ddsnap.c - zumastor - Google Code

‹r1724

r1735

Source path: svn/ trunk/ ddsnap/ ddsnap.c

#define _GNU_SOURCE // for O_DIRECT

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
#include <string.h>
#include <fcntl.h>
#include <errno.h>
#include <inttypes.h>
#include <netinet/in.h>
#include <popt.h>
#include <time.h>
#include <sys/time.h>
#include <ctype.h>
#include <zlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <sys/un.h>
#include <linux/fs.h> // for BLKGETSIZE
#include <poll.h>
#include <sys/prctl.h>
#include "dm-ddsnap.h"
#include "buffer.h"
#include "daemonize.h"
#include "ddsnap.h"
#include "ddsnap.agent.h"
#include "delta.h"
#include "diskio.h"
#include "list.h"
#include "sock.h"
#include "trace.h"
#include "build.h"

/* Utilities */

// FIXME: Overloading *_MAX is bad
#define INPUT_ERROR UINT32_MAX
#define INPUT_ERROR_64 UINT64_MAX
#define bogus 0 // suppress uninitialized variable warning, probably indicates of poor algorithm

u64 strtobytes64(char const *string)
{
	unsigned long long bytes = 0;
	char *letter = NULL;

	bytes = strtoull(string, &letter, 10);

	if (bytes < 0)
		return INPUT_ERROR_64;

	if (bytes == ULLONG_MAX && errno == ERANGE)
		return INPUT_ERROR_64;

	if (letter[0] == '\0')
		return bytes;

	if (letter[1] != '\0')
		return INPUT_ERROR_64;

	switch (letter[0]) {
	case 'k': case 'K':
		return bytes * (1 << 10);
	case 'm': case 'M':
		return bytes * (1 << 20);
	case 'g': case 'G':
		return bytes * (1 << 30);
	default:
		return INPUT_ERROR_64;
	}
}

u32 strtobytes(char const *string)
{
	unsigned long long bytes = 0;

	bytes = strtobytes64(string);
	if (bytes > ULONG_MAX || bytes == INPUT_ERROR_64)
		return INPUT_ERROR;
	return (u32)bytes;
}

u32 strtobits(char const *string)
{
	long amount = 0;
	u32 bits = 0;
	char *letter = NULL;

	amount = strtol(string, &letter, 10);

	if ((amount <= 0) || (amount & (amount - 1)))
		return INPUT_ERROR;

	while (amount > 1) {
		bits += 1;
		amount >>= 1;
	}

	switch (letter[0]) {
	case '\0':
		break;
	case 'k': case 'K':
		bits += 10;
		break;
	case 'm': case 'M':
		bits += 20;
		break;
	case 'g': case 'G':
		bits += 30;
		break;
	default:
		return INPUT_ERROR;
		break;
	}

	if (letter[1] != '\0')
		return INPUT_ERROR;

	return bits;
}

char reason[200]; /* detailed error message: nonnull implies nonzero errno */

static void errprint(char *action)
{
	fprintf(stderr, "could not %s %s%s (%s)\n", action, strlen(reason) ? "because " : "", reason, strerror(errno));
}

static int discard(int fd, int len)
{
	int i, err;

	for (i = 0; i < len; i++)
		if ((err = fdread(fd, (char[1]){ }, 1)) < 0) {
			errno = -err;
			return -1;
		}
	return 0;
}

/* changelist and delta file header info */

#define MAGIC_SIZE 8
#define CHANGELIST_MAGIC_ID "rln"
#define DELTA_MAGIC_ID "jc"
#define MAGIC_NUM 0xbead0023

#define DEFAULT_REPLICATION_PORT 4321
#define TRUE 1
#define FALSE 0

#define XDELTA 1
#define RAW (1 << 1)
#define BEST_COMP (1 << 2)

#define DEF_GZIP_COMP 0
#define MAX_GZIP_COMP 9

#define MAX_MEM_BITS 20
#define MAX_MEM_SIZE (1 << MAX_MEM_BITS)
#define DEFAULT_CHUNK_SIZE_BITS 12
#define DEFAULT_JOURNAL_SIZE (1000 * SECTOR_SIZE)

struct cl_header
{
	char magic[MAGIC_SIZE];
	u32 chunksize_bits;
	u32 src_snap;
	u32 tgt_snap;
};

struct delta_header
{
	char magic[MAGIC_SIZE];
	u64 chunk_num;
	u32 chunk_size;
	u32 src_snap;
	u32 tgt_snap;
} PACKED;

struct delta_extent_header
{
	u32 magic_num;
	u32 mode;
	u32 gzip_on;
	u64 extent_addr;
	u64 num_of_chunks;
	u64 extents_delta_length;
	u64 ext1_chksum;
	u64 ext2_chksum;
} PACKED;

static u64 checksum(const unsigned char *data, u32 data_length)
{
	u64 result = 0;
	u32 i;

	for (i = 0; i < data_length; i++)
		result = result + data[i];


	return result;
}

static int read_reason(int sock, int size)
{
	int some = size;
	if (some >= sizeof(reason))
		some = sizeof(reason) - 1;
	read(sock, reason, some);
	reason[some] = 0;
	if (size > some) {
		warn("incoming error message too long, %i bytes", size);
		discard(sock, size - some);
	}
	return -1;
}

static int unknown_message(int sock, struct head *head)
{
	int err;
	if (head->code == PROTOCOL_ERROR) {
		struct protocol_error error;
		if (head->length < sizeof(error)) {
			snprintf(reason, sizeof(reason),
				"expected protocol error of %zu bytes but only got %u",
					 sizeof(error), head->length);
			err = EINVAL;
			return -1;
		}
		if ((err = readpipe(sock, &error, sizeof(error))) < 0) {
			errno = -err;
			snprintf(reason, sizeof(reason),
				"unable to receive protocol error message");
			return -1;
		}
		return read_reason(sock, head->length - sizeof(error));
	}
	if (discard(sock, head->length))
		return -1;
	snprintf(reason, sizeof(reason),
		"received unknown message, code %x, size %i", head->code, head->length);
	errno = EINVAL;
	return -1;
}

static int generic_error(int sock, struct head *head)
{
	if (head->code == GENERIC_ERROR) {
		struct generic_error error;
		if (read(sock, &error, sizeof(error)) == -1)
			return -1;
		errno = error.err;
		return read_reason(sock, head->length - sizeof(error));
	}
	return unknown_message(sock, head);
}

static int expect(int sock, int code)
{
	struct head head;
	int err;

	if ((err = readpipe(sock, &head, sizeof(head))) < 0) {
		strcpy(reason, "unable to read message header");
		errno = -err;
		return -1;
	}
	if (head.code != code)
		return generic_error(sock, &head);
	return head.length;
}

static int get_reply(int sock, char *request, int code_ok, int reply_len, void *reply)
{
	int err, size;

	if ((size = expect(sock, code_ok)) == -1)
		return -1;
	if (size != reply_len) {
		snprintf(reason, sizeof(reason),
			"%s length mismatch: expected >=%i, actual %i", request, reply_len, size);
		errno = EINVAL;
		return -1;
	}
	if ((err = readpipe(sock, reply, size)) < 0) {
		errno = -err;
		snprintf(reason, sizeof(reason), "failed to receive message");
		return -1;
	}
	return 0;
}

static char *get_message(int sock, unsigned int length) // !!! kill me
{
	char *buffer;
	if (!(buffer = malloc(length)))
		return NULL;

	int err;
	if ((err = readpipe(sock, buffer, length)) < 0) {
		warn("unable to get reason why command failed: %s", strerror(err));
		free(buffer);
		return NULL;
	}
	buffer[length-1] = '\0';

	return buffer;
}

static void error_message_handler(int sock, const char *prefix, unsigned int length) // !!! kill me
{
	char *buffer = get_message(sock, length);
	warn("%s : %s", prefix, (buffer? buffer : "no error message"));
	if (buffer)
		free(buffer);
}

static int create_socket(char const *sockname)
{
	struct sockaddr_un addr = { .sun_family = AF_UNIX };
	int addr_len = sizeof(addr) - sizeof(addr.sun_path) + strlen(sockname);
	int sock;

	if ((sock = socket(AF_UNIX, SOCK_STREAM, 0)) == -1)
		error("Can't get AF_UNIX socket: %s", strerror(errno));
	strncpy(addr.sun_path, sockname, sizeof(addr.sun_path));
	if (sockname[0] == '@')
		addr.sun_path[0] = 0;

	if (connect(sock, (struct sockaddr *)&addr, addr_len) == -1)
		error("Can't connect to control socket %s: %s", sockname, strerror(errno));

	return sock;
}

static int parse_snaptag(char const *snapstr, u32 *snaptag)
{
	if (snapstr[0] == '\0' || snapstr[0] == '-')
		return -1;

	unsigned long num;
	char *endptr;

	num = strtoul(snapstr, &endptr, 10);

	if (*endptr != '\0')
		return -1;

	if (num >= ULONG_MAX)
		return -1;

	*snaptag = num;
	return 0;
}

#define CHUNK_ARRAY_INIT 1024

struct change_list *init_change_list(u32 chunksize_bits, u32 src_snap, u32 tgt_snap)
{
	struct change_list *cl;

	if ((cl = malloc(sizeof(struct change_list))) == NULL)
		return NULL;

	cl->count = 0;
	cl->length = CHUNK_ARRAY_INIT;
	cl->chunksize_bits = chunksize_bits;
	cl->src_snap = src_snap;
	cl->tgt_snap = tgt_snap;
	cl->chunks = malloc(cl->length * sizeof(u64));

	if (cl->chunks == NULL) {
		free(cl);
		return NULL;
	}

	return cl;
}

int append_change_list(struct change_list *cl, u64 chunkaddr)
{
	if ((cl->count + 1) >= cl->length) {
		u64 *newchunks;

		/* we could use a different strategy here like incrementing but this
		 * should work and will avoid having too many
		 * reallocations
		 */
		if ((newchunks = realloc(cl->chunks, cl->length * 2 * sizeof(u64))) == NULL)
		    return -1;

		cl->length *= 2;
		cl->chunks = newchunks;
	}

	cl->chunks[cl->count] = chunkaddr;
	cl->count++;

	return 0;
}


void free_change_list(struct change_list *cl)
{
	if (cl->chunks)
		free(cl->chunks);
	free(cl);
}

static struct change_list *read_changelist(int cl_fd)
{
	struct cl_header clh;

	if (fdread(cl_fd, &clh, sizeof(clh)) < 0) {
		warn("Not a proper changelist file (too short for header)");
		return NULL;
	}

	if (strncmp(clh.magic, CHANGELIST_MAGIC_ID, MAGIC_SIZE) != 0) {
		warn("Not a proper changelist file (wrong magic in header: %s)", clh.magic);
		return NULL;
	}

	struct change_list *cl;

	if ((cl = init_change_list(clh.chunksize_bits, clh.src_snap, clh.tgt_snap)) == NULL)
		return NULL;

	u64 chunkaddr = 0;
	ssize_t len;

	trace_on(printf("reading changelist for snapshots %u and %u from file\n", clh.src_snap, clh.tgt_snap););
	for (;;) {
		len = read(cl_fd, &chunkaddr, sizeof(u64)); /* FIXME: need to handle short read but fdread() doesn't distinguish short read vs. no read EOF */

		if (len == 0)
			break;

		if (len != sizeof(u64)) {
			error("Incomplete chunk address.");
			break;
		}

		if (chunkaddr == -1)
			break;

		if (append_change_list(cl, chunkaddr) < 0)
			error("unable to append chunk address.");
	}

	if (chunkaddr != -1)
		warn("changelist file may be incomplete");

	trace_on(printf("done reading "U64FMT" chunk addresses\n", cl->count););

	return cl;
}

static int write_changelist(int change_fd, struct change_list const *cl)
{
	struct cl_header clh;
	int err;

	strncpy(clh.magic, CHANGELIST_MAGIC_ID, sizeof(clh.magic));
	clh.chunksize_bits = cl->chunksize_bits;
	clh.src_snap = cl->src_snap;
	clh.tgt_snap = cl->tgt_snap;

	if ((err = fdwrite(change_fd, &clh, sizeof(clh))) < 0) {
		warn("unable to write magic information to changelist file: %s", strerror(-err));
		return -1;
	}

	if ((err = fdwrite(change_fd, cl->chunks, cl->count * sizeof(cl->chunks[0]))) < 0) {
		warn("unable to write changelist file: %s", strerror(-err));
		return -1;
	}

	u64 marker = -1;

	if ((err = fdwrite(change_fd, &marker, sizeof(marker))) < 0) {
		warn("unable to write changelist marker: %s", strerror(-err));
		return err;
	}

	return 0;
}

static u64 chunks_in_extent(struct change_list *cl, u64 pos, u32 chunk_size)
{
	u64 start_chunkaddr, cur_chunkaddr, num_of_chunks = 1;
	start_chunkaddr = cl->chunks[pos];
	cur_chunkaddr = cl->chunks[pos+1];

	while (cur_chunkaddr == start_chunkaddr + num_of_chunks && (num_of_chunks < (MAX_MEM_SIZE / chunk_size))) {
		num_of_chunks++;
		cur_chunkaddr = cl->chunks[pos + num_of_chunks];
	}
	return num_of_chunks;
}

static int create_raw_delta(struct delta_extent_header *deh_ptr, const unsigned char *input_buffer, unsigned char *output_buffer, u64 input_size, u64 *output_size) 
{
	memcpy(output_buffer, input_buffer, input_size);
	*output_size = input_size;
	deh_ptr->mode = RAW;
	deh_ptr->extents_delta_length = input_size;
	return 0;
}

static int create_xdelta_delta(struct delta_extent_header *deh_ptr, unsigned char *input_buffer1, unsigned char *input_buffer2, unsigned char *output_buffer, u64 input_size, u64 *output_size)
{
	trace_off(printf("create xdelta delta\n"););
	int err;
	int ret = create_delta_chunk(input_buffer1, input_buffer2, output_buffer, input_size, (int *)output_size);
	deh_ptr->mode = XDELTA;
	deh_ptr->extents_delta_length = *output_size;

	/* If xdelta result is larger than or equal to extent_size, we want to just copy over the raw extent */
	if (ret == BUFFER_SIZE_ERROR || *output_size == input_size) {
		trace_off(printf("extents_delta size is larger than or equal to extent_size\n"););
		create_raw_delta(deh_ptr, input_buffer2, output_buffer, input_size, output_size);
	} else if (ret < 0) {
		goto gen_create_error;
	} else if (ret >= 0) {
		/* sanity test for xdelta creation */
		unsigned char *delta_test = malloc(input_size);
		if (!delta_test)
			goto gen_alloc_error;

		ret = apply_delta_chunk(input_buffer1, delta_test, output_buffer, input_size, *output_size);

		if (ret != input_size) {
			free(delta_test);
			goto gen_applytest_error;
		}

		if (memcmp(delta_test, input_buffer2, input_size) != 0) {
			trace_off(printf("generated delta does not match extent on disk.\n"););
			create_raw_delta(deh_ptr, input_buffer2, output_buffer, input_size, output_size);
		}
		trace_off(printf("able to generate delta\n"););
		free(delta_test);
	}
	return 0;

gen_create_error:
	err = -ERANGE;
	warn("unable to create delta: %s", strerror(-err));
	return err;

gen_alloc_error:
	err = -ENOMEM;
	warn("memory allocation failed: %s", strerror(-err));
	return err;

gen_applytest_error:
	err = -ERANGE;
	warn("test application of delta failed: %s", strerror(-err));
	return err;
}

static int gzip_on_delta(struct delta_extent_header *deh_ptr, unsigned char *input_buffer, unsigned char *output_buffer, u64 input_size, u64 *output_size, int level)
{
	int err;
	*output_size = input_size + 12 + (input_size >> 9);
	int comp_ret = compress2(output_buffer, (unsigned long *) output_size, input_buffer, input_size, level);

	if (comp_ret == Z_MEM_ERROR)
		goto gen_compmem_error;
	if (comp_ret == Z_BUF_ERROR)
		goto gen_compbuf_error;
	if (comp_ret == Z_STREAM_ERROR)
		goto gen_compstream_error;

	if (*output_size < input_size) {
		deh_ptr->gzip_on = TRUE;
		deh_ptr->extents_delta_length = *output_size;
	} else {
		deh_ptr->gzip_on = FALSE;
		deh_ptr->extents_delta_length = input_size;
		memcpy(output_buffer, input_buffer, input_size);
		*output_size = input_size;
	}
	return 0;

gen_compmem_error:
	err = -ENOMEM;
	warn("not enough buffer memory for compression of delta: %s", strerror(-err));
	return err;

gen_compbuf_error:
	err = -ERANGE;
	warn("not enough room in the output buffer for compression of delta: %s", strerror(-err));
	return err;

gen_compstream_error:
	err = -ERANGE;
	warn("invalid compression parameter level=%d delta_size=%Lu in delta: %s", level, input_size, strerror(-err));
	return err;
}


static struct status_reply *generate_status(int serv_fd, u32 snaptag)
{
	int err, size;
	struct status_reply *reply;

	if ((err = outbead(serv_fd, STATUS, struct status_request, snaptag))) {
		warn("unable to send status request: %s", strerror(-err));
		return NULL;
	}

	if ((size = expect(serv_fd, STATUS_OK)) == -1) {
		errprint("get status");
		return NULL;
	}

	if (size < sizeof(struct status_reply)) {
		warn("status length mismatch: expected >=%zu, actual %u", sizeof(struct status_reply), size);
		return NULL;
	}

	if (!(reply = malloc(size))) {
		warn("unable to allocate %Lu bytes for reply buffer", (long long)size);
		return NULL;
	}

	/* We won't bother to check that the lengths match because it would
	 * be ugly to read in the structure in pieces.
	 */
	if ((err = readpipe(serv_fd, reply, size)) < 0) {
		warn("received incomplete status message: %s", strerror(-err));
		free(reply);
		return NULL;
	}

	return reply;
}

/* returns the current time in seconds since the epoch */
int now(void) {
	struct timeval tv;

	gettimeofday(&tv, NULL);
	return tv.tv_sec;
}

static char *malloc_snapshot_name(const char *devstem, u32 id)
{
	char *snapshot_name = NULL;
	int length = strlen(devstem) + 32; // !!! why 32?

	if (!(snapshot_name = malloc(length))) {
		warn("unable to allocate memory snapshot name");
		return NULL;
	}

	snprintf(snapshot_name, length, "%s(%Lu)", devstem, (llu_t) id);

	return snapshot_name;
}

static int generate_progress_file(const char *progress_file, char **tmpfilename)
{
	char *progress_tmpfile;
	int progress_len = strlen(progress_file);
	if (!(progress_tmpfile = malloc(progress_len + 5))) {
		warn("unable to allocate progress tempfile name");
		return -ENOMEM;
	}
	strncpy(progress_tmpfile, progress_file, progress_len);
	strncpy(progress_tmpfile + progress_len, ".tmp", 5);
	*tmpfilename = progress_tmpfile;
	return 0;
}

static int write_progress(const char *progress_file, const char *progress_tmpfile, u64 chunk_num, u64 chunk_count, u64 extent_addr, u32 tgt_snap)
{
	FILE *progress_fs;
	if ((progress_fs = fopen(progress_tmpfile, "w")) == NULL) {
		warn("unable to open progress temp file %s: %s", progress_tmpfile, strerror(errno));
		return -1;
	}
	if (fprintf(progress_fs, "%u %Lu/%Lu %Lu\n", tgt_snap, chunk_num, chunk_count, extent_addr) < 0) {
		warn("unable write to progress temp file %s: %s", progress_tmpfile, strerror(errno));
		fclose(progress_fs);
		return -1;
	}
	fclose(progress_fs);
	if (rename(progress_tmpfile, progress_file) < 0) {
		warn("unable to rename progress file: %s", strerror(errno));
		return -1;
	}
	return 0;
}

/* returns the current time in useconds since the epoch */
u64 usec_now(void) {
	struct timeval tv;

	gettimeofday(&tv, NULL);
	return (tv.tv_sec * 1000000) + tv.tv_usec;
}

static void usec_sleep(unsigned long sleep_usecs)
{
	struct timespec sleep_time, left_time;
	sleep_time.tv_sec = sleep_usecs / 1000000;
	sleep_time.tv_nsec = (sleep_usecs % 1000000) * 1000;
	while (nanosleep(&sleep_time, &left_time) && (errno == EINTR)) {
		sleep_time.tv_sec = left_time.tv_sec;
		sleep_time.tv_nsec = left_time.tv_nsec;
	}
}
static int generate_delta_extents(u32 mode, int level, struct change_list *cl, int deltafile, char const *devstem, u32 src_snap, u32 tgt_snap, char const *progress_file, u64 start_chunk, u32 rate_limit)
{
	int fullvolume = (src_snap == -1);
	char *dev1name = NULL, *dev2name = NULL, *progress_tmpfile = NULL;
	int snapdev1 = -1, snapdev2 = -1;
	int err = -ENOMEM;
	unsigned char *dev1_extent   = malloc(MAX_MEM_SIZE);
	unsigned char *dev2_extent   = malloc(MAX_MEM_SIZE);
	unsigned char *extents_delta = malloc(MAX_MEM_SIZE);
	unsigned char *gzip_delta    = malloc(MAX_MEM_SIZE + 12 + (MAX_MEM_SIZE >> 9));
	unsigned char *dev2_gzip_extent = malloc(MAX_MEM_SIZE + 12 + (MAX_MEM_SIZE >> 9));

	if (!fullvolume && (!(dev1name = malloc_snapshot_name(devstem, src_snap)) || ((snapdev1 = open(dev1name, O_RDONLY)) < 0))) {
		warn("unable to open source snapshot: %s", strerror(errno));
		goto out;
	}
	if (!(dev2name = malloc_snapshot_name(devstem, tgt_snap)) || ((snapdev2 = open(dev2name, O_RDONLY)) < 0)) {
		warn("unable to open target snapshot: %s", strerror(errno));
		goto out;
	}

	if (!dev1_extent || !dev2_extent || !extents_delta || !gzip_delta || !dev2_gzip_extent) {
		warn("variable memory allocation failed: %s", strerror(-err));
		goto out;
	}

	u64 dev2_gzip_size;
	struct delta_extent_header deh = { .magic_num = MAGIC_NUM };
	u64 extent_addr = bogus, chunk_num, num_of_chunks, source_volume_size = bogus, target_volume_size;
	u64 extent_size, delta_size, gzip_size = MAX_MEM_SIZE, bytes_total = 0, bytes_sent = 0;
	u32 chunk_size = 1 << cl->chunksize_bits;
	struct delta_extent_header deh2 = { .magic_num = MAGIC_NUM, .mode = RAW };

	trace_off(printf("dev1name: %s, dev2name: %s\n", dev1name, dev2name););
	trace_off(printf("level: %d, chunksize bits: %Lu, chunk_count: %Lu\n", level, (llu_t) cl->chunksize_bits, (llu_t) cl->count););
	trace_off(printf("starting delta generation, mode %u, chunksize %u\n", mode, chunk_size););

	if (!fullvolume && (source_volume_size = fdsize64(snapdev1)) == -1) {
		warn("unable to determine volume size for %s", dev1name);
		goto out;
	}
	if ((target_volume_size = fdsize64(snapdev2)) == -1) {
		warn("unable to determine volume size for %s", dev2name);
		goto out;
	}
	if (progress_file && (err = generate_progress_file(progress_file, &progress_tmpfile)))
		goto out;

	u64 current_time, last_update = 0, start_time = usec_now();

	for (chunk_num = start_chunk; chunk_num < cl->count;) {
		if (fullvolume) {
			extent_size = chunk_size;
			extent_addr = chunk_num * extent_size;
			num_of_chunks = 1;
		} else {
			extent_addr = cl->chunks[chunk_num] << cl->chunksize_bits;
			if (chunk_num == (cl->count - 1) )
				num_of_chunks = 1;
			else
				num_of_chunks = chunks_in_extent(cl, chunk_num, chunk_size);
			extent_size = chunk_size * num_of_chunks;
		}
		if (extent_addr > target_volume_size - extent_size)
			extent_size = target_volume_size - extent_addr;
		bytes_total += extent_size;

		/* delta extent header set-up*/
		deh.gzip_on = FALSE;
		deh.extent_addr = extent_addr;
		deh.num_of_chunks = num_of_chunks;

		/* read in extents from dev1 & dev2 */
		if (!fullvolume && source_volume_size > extent_addr) {
			/* deal with the last extent of the source snapshot */
			u64 source_extent_size = (extent_addr > source_volume_size - extent_size) ? (source_volume_size - extent_addr) : extent_size;
			if ((err = diskread(snapdev1, dev1_extent, source_extent_size, extent_addr)) < 0) {
				warn("read from snapshot device \"%s\" failed ", dev1name);
				goto error_source;
			}
			if (posix_fadvise(snapdev1, extent_addr, source_extent_size, POSIX_FADV_DONTNEED) != 0)
				warn("can't free cached pages for the source snapshot, error %s", strerror(errno));
			deh.ext1_chksum = checksum((const unsigned char *) dev1_extent, source_extent_size);
		}
		if ((err = diskread(snapdev2, dev2_extent, extent_size, extent_addr)) < 0) {
			warn("read from snapshot device \"%s\" failed ", dev2name);
			goto error_source;
		}
		if (posix_fadvise(snapdev2, extent_addr, extent_size, POSIX_FADV_DONTNEED) != 0)
			warn("can't free cached pages for the target snapshot, error %s", strerror(errno));
		deh.ext2_chksum = checksum((const unsigned char *) dev2_extent, extent_size);


		if (fullvolume || (extent_addr > source_volume_size - extent_size)) {
			/* copy RAW data of snap2 if it is fullvolume or if snap2 is larger than snap1 */
			deh.extents_delta_length = extent_size;
			deh.mode = RAW;
			if ((err = gzip_on_delta(&deh, dev2_extent, gzip_delta, extent_size, &gzip_size, level)) < 0)
				goto error_source;
		} else {
			/* Three different modes, raw, xdelta, best (either gzipped raw or gzipped xdelta)
			 * Always use raw when senind the first chunk we are resuming on */
			if (mode == RAW)
				err = create_raw_delta (&deh, dev2_extent, extents_delta, extent_size, &delta_size);
			else // compute xdelta for XDELTA or BEST_COMP mode
				err = create_xdelta_delta (&deh, dev1_extent, dev2_extent, extents_delta, extent_size, &delta_size);
			if ((err < 0) || ((err = gzip_on_delta(&deh, extents_delta, gzip_delta, delta_size, &gzip_size, level)) < 0))
				goto error_source;

			if (mode == BEST_COMP) {
				/* delta extent header set-up for dev2_extent */
				deh2.gzip_on = FALSE;
				deh2.extents_delta_length = extent_size;
				if ((err = gzip_on_delta(&deh2, dev2_extent, dev2_gzip_extent, extent_size, &dev2_gzip_size, level)) < 0)
					goto error_source;
				if (dev2_gzip_size <= gzip_size) {
					deh.mode = deh2.mode;
					deh.gzip_on = deh2.gzip_on;
					deh.extents_delta_length = deh2.extents_delta_length;
					memcpy(gzip_delta, dev2_gzip_extent, dev2_gzip_size);
				}
			}
		}

		/* write the delta extent header and extents_delta to the delta file*/
		if ((err = fdwrite(deltafile, &deh, sizeof(deh))) < 0) {
			warn("unable to write delta header ");
			goto error_source;
		}
		if ((err = fdwrite(deltafile, gzip_delta, deh.extents_delta_length)) < 0) {
			warn("unable to write delta data ");
			goto error_source;
		}
		bytes_sent += deh.extents_delta_length + sizeof(deh);

		current_time = usec_now();
		if (rate_limit && ((bytes_sent * 1000000 / rate_limit) > (current_time - start_time)))
			usec_sleep(bytes_sent * 1000000 / rate_limit - (current_time - start_time));

		if (progress_file && ((current_time - last_update) > 1000000)) {
			if (write_progress(progress_file, progress_tmpfile, chunk_num, cl->count, extent_addr, tgt_snap) < 0)
				goto out;
			last_update = current_time;
		}

		chunk_num = chunk_num + num_of_chunks;
	}

	/* Make sure everything in changelist was properly transmitted */
	if (chunk_num != cl->count) {
		warn("changelist was not fully transmitted");
		err = -ERANGE;
	} else {
		current_time = usec_now();
		u32 transrate = (current_time > start_time) ? (unsigned)(bytes_sent * 1000000 / (current_time - start_time)) : 0;
		warn("Total chunks %Lu (%Lu bytes), wrote %Lu bytes in %i seconds, rate limit %u, transfer rate %u bytes/s", chunk_num, bytes_total, bytes_sent, (unsigned)((current_time - start_time) / 1000000), rate_limit, transrate);
		err = progress_file ? write_progress(progress_file, progress_tmpfile, chunk_num, cl->count, extent_addr, tgt_snap) : 0;
	}
	goto out;
error_source:
	warn("for "U64FMT" chunk extent starting at offset "U64FMT": %s", num_of_chunks, extent_addr, strerror(-err));
out:
	if (snapdev1 >= 0)
		close(snapdev1);
	if (snapdev2 >= 0)
		close(snapdev2);
	if (dev1_extent)
		free(dev1_extent);
	if (dev2_extent)
		free(dev2_extent);
	if (extents_delta)
		free(extents_delta);
	if (gzip_delta)
		free(gzip_delta);
	if (dev2_gzip_extent)
		free(dev2_gzip_extent);
	if (progress_tmpfile)
		free(progress_tmpfile);
	if (dev1name)
		free(dev1name);
	if (dev2name)
		free(dev2name);
	return err;
}

static int generate_delta(u32 mode, int level, struct change_list *cl, int deltafile, char const *devstem)
{
	/* Delta header set-up */
	struct delta_header dh;

	strncpy(dh.magic, DELTA_MAGIC_ID, sizeof(dh.magic));
	dh.chunk_num = cl->count;
	dh.chunk_size = 1 << cl->chunksize_bits;
	dh.src_snap = cl->src_snap;
	dh.tgt_snap = cl->tgt_snap;

	trace_off(fprintf(stderr, "writing delta file with chunk_num=%Lu chunk_size=%Lu mode=%Lu\n", (llu_t) dh.chunk_num, (llu_t) dh.chunk_size, (llu_t) mode););

	int err;
	if ((err = fdwrite(deltafile, &dh, sizeof(dh))) < 0)
		return err;

	return generate_delta_extents(mode, level, cl, deltafile, devstem, dh.src_snap, dh.tgt_snap, NULL, 0, 0);
}

static int ddsnap_generate_delta(u32 mode, int level, char const *changelistname, char const *deltaname, char const *devstem)
{
	int clfile = open(changelistname, O_RDONLY);
	if (clfile < 0) {
		warn("could not open changelist file \"%s\" for reading: %s", changelistname, strerror(errno));
		return 1;
	}

	struct change_list *cl;

	if ((cl = read_changelist(clfile)) == NULL) {
		warn("unable to parse changelist file \"%s\"", changelistname);
		close(clfile);
		return 1;
	}

	close(clfile);

	int deltafile = open(deltaname, O_CREAT|O_WRONLY|O_TRUNC, S_IRWXU);
	if (deltafile < 0) {
		warn("could not create delta file \"%s\": %s", deltaname, strerror(errno));
		free_change_list(cl);
		return 1;
	}

	if (generate_delta(mode, level, cl, deltafile, devstem) < 0) {
		warn("could not write delta file \"%s\"", deltaname);
		close(deltafile);
		free_change_list(cl);
		return 1;
	}

	close(deltafile);
	free_change_list(cl);

	return 0;
}

static struct change_list *stream_changelist(int serv_fd, u32 src_snap, u32 tgt_snap)
{
	int err;

	if ((err = outbead(serv_fd, STREAM_CHANGELIST, struct stream_changelist, src_snap, tgt_snap))) {
		warn("could not request change list: %s", strerror(-err));
		return NULL;
	}

 	struct changelist_stream cl_head;
	if (get_reply(serv_fd, "change list", STREAM_CHANGELIST_OK, sizeof(struct changelist_stream), &cl_head) != 0) {
		errprint("request change list");
		return NULL;
	}

	struct change_list *cl;

	if ((cl = malloc(sizeof(struct change_list))) == NULL) {
		/* FIXME: need to read the data anyway to clear the socket */
		warn("unable to allocate change list");
		return NULL;
	}

	cl->count = cl_head.chunk_count;
	cl->chunksize_bits = cl_head.chunksize_bits;
	cl->src_snap = src_snap;
	cl->tgt_snap = tgt_snap;

	if (cl->chunksize_bits == 0) {
		/* FIXME: need to read the data anyway to clear the socket */
		warn("invalid chunk size %Lu in REPLY_STREAM_CHANGE_LIST", (llu_t) cl->chunksize_bits);
		free(cl);
		return NULL;
	}

	if (cl->count == 0) {
		cl->chunks = NULL;
		return cl;
	}

	if ((cl->chunks = malloc(cl->count * sizeof(cl->chunks[0]))) == NULL) {
		/* FIXME: need to read the data anyway to clear the socket */
		warn("unable to allocate change list chunks");
		free(cl);
		return NULL;
	}

	trace_off(printf("reading %Lu chunk addresses (%Lu bits) from ddsnapd\n", (llu_t) cl->count, (llu_t) cl->chunksize_bits););
	if ((err = readpipe(serv_fd, cl->chunks, cl->count * sizeof(cl->chunks[0]))) < 0) {
		warn("unable to read change list chunks: %s", strerror(-err));
		free(cl->chunks);
		free(cl);
		return NULL;
	}


	return cl;
}

static u64 get_snapshot_sectors(int serv_fd, u32 snaptag)
{
	int err;

	if ((err = outbead(serv_fd, REQUEST_SNAPSHOT_SECTORS, struct status_request, snaptag))) {
		warn("unable to send request for snapshot sector count: %s", strerror(-err));
		return 0;
	}

	struct snapshot_sectors reply;
	if (get_reply(serv_fd, "snapshot_sectors", SNAPSHOT_SECTORS, sizeof(struct snapshot_sectors), &reply) != 0) {
		errprint("get snapshot sectors");
		return 0;
	}
	return reply.count;
}

static int ddsnap_replication_send(int serv_fd, u32 src_snap, u32 tgt_snap, char const *devstem, u32 mode, int level, int ds_fd, char const *progress_file, u64 start_addr, u32 ratelimit)
{
	int fullvolume = (src_snap == -1), err = -ENOMEM;
	u64 skip_chunks = 0;
	struct change_list *cl = NULL;

	/* setup changelist */
	if (fullvolume) {
		if ((cl = malloc(sizeof(struct change_list))) == NULL) {
			warn("unable to allocate change list");
			goto out;
		}

		err = -EINVAL;
		struct status_reply *reply;
		if (!(reply = generate_status(serv_fd, ~((u32)0U)))) {
			warn("cannot generate status");
			goto out;
		}
		cl->chunksize_bits = reply->meta.chunksize_bits;
		free(reply);

		u64 vol_size_bytes = get_snapshot_sectors(serv_fd, (u32)~0UL) * 512, chunkmask = ~(-1 << cl->chunksize_bits);
		cl->count = (vol_size_bytes >> cl->chunksize_bits) + !!(vol_size_bytes & chunkmask);
		cl->length = CHUNK_ARRAY_INIT;
		cl->src_snap = src_snap;
		cl->tgt_snap = tgt_snap;
		cl->chunks = NULL;
		skip_chunks = cl->count - ((vol_size_bytes - start_addr) >> cl->chunksize_bits);
	} else {
		trace_off(printf("requesting changelist from snapshot %Lu to %Lu\n", (llu_t) src_snap, (llu_t) tgt_snap););
		if ((cl = stream_changelist(serv_fd, src_snap, tgt_snap)) == NULL) {
			warn("could not receive change list for snapshots %Lu and %Lu", (llu_t) src_snap, (llu_t) tgt_snap);
			err = -EINVAL;
			goto out;
		}
		/* If resuming, find the first chunk that starts after the start_addr, TODO binary search */
		while (skip_chunks < cl->count && cl->chunks[skip_chunks] << cl->chunksize_bits < start_addr)
			skip_chunks++;
	}

	/* request approval for delta send */
	if ((err = outbead(ds_fd, SEND_DELTA, struct delta_header, DELTA_MAGIC_ID, cl->count - skip_chunks, 1 << cl->chunksize_bits, src_snap, tgt_snap))) {
		warn("unable to send delta: %s", strerror(-err));
		goto out;
	}

	struct head head;
	if ((err = readpipe(ds_fd, &head, sizeof(head))) < 0) {
		warn("unable to read response from downstream: %s", strerror(-err));
		goto out;
	}

	if (head.code != SEND_DELTA_PROCEED) {
		if (head.code != SEND_DELTA_ERROR)
			unknown_message(ds_fd, &head);
		else
			error_message_handler(ds_fd, "downstream server reason why send delta failed", head.length);
		err = -EPIPE;
		goto out;
	}

	warn("sending delta from %i to %i", src_snap, tgt_snap);

	/* stream delta */
	if ((err = generate_delta_extents(mode, level, cl, ds_fd, devstem, src_snap, tgt_snap, progress_file, skip_chunks, ratelimit)) < 0) {
		warn("could not send delta downstream for snapshots %i and %i", src_snap, tgt_snap);
		goto out;
	}
	if ((err = readpipe(ds_fd, &head, sizeof(head))) < 0) {
		warn("unable to read response from downstream: %s", strerror(-err));
		goto out;
	}

	if (head.code != SEND_DELTA_DONE) {
		if (head.code != SEND_DELTA_ERROR)
			unknown_message(ds_fd, &head);
		else
			error_message_handler(ds_fd, "downstream server reason why send delta failed", head.length);
		err = -EPIPE;
		goto out;
	}
	err = 0;
out:
	if (cl)
		free_change_list(cl);
	return err;
}

static int apply_delta_extents(int deltafile, u32 chunk_size, u64 chunk_count, char const *dev1name, char const *dev2name, char const *progress_file, u32 tgt_snap)
{
	int fullvolume = !dev1name;
	int snapdev1 = bogus, snapdev2;
	int err = bogus;
	unsigned char *updated=NULL, *extent_data=NULL, *delta_data=NULL, *comp_delta=NULL;
	char *up_extent1=NULL, *up_extent2=NULL;

	/* if an extent is being applied */
	if (!fullvolume && ((snapdev1 = open(dev1name, O_RDONLY)) < 0)) {
		warn("could not open snapdev file \"%s\" for reading: %s.", dev1name, strerror(errno));
		return -errno;
	}

	if ((snapdev2 = open(dev2name, O_WRONLY)) < 0) {
		warn("could not open snapdev file \"%s\" for writing: %s.", dev2name, strerror(errno));
		if (!fullvolume)
			close(snapdev1);
		return -errno;
	}

	char *progress_tmpfile = NULL;
	if (progress_file && (err = generate_progress_file(progress_file, &progress_tmpfile)))
		goto out;

	if (!(updated = malloc(MAX_MEM_SIZE)) || !(extent_data = malloc(MAX_MEM_SIZE)) \
		|| !(delta_data = malloc(MAX_MEM_SIZE)) || !(comp_delta = malloc(MAX_MEM_SIZE)) \
		|| !(up_extent1 = malloc(MAX_MEM_SIZE)) || !(up_extent2 = malloc(MAX_MEM_SIZE))) {
		warn("memory allocation failed: %s", strerror(errno));
		err = -ENOMEM;
		goto out;
	}

	struct delta_extent_header deh;
	u64 uncomp_size, extent_size, source_volume_size = bogus, target_volume_size;
	u64 extent_addr = 0, chunk_num;
	int current_time, last_update = 0;

	if (!fullvolume && (source_volume_size = fdsize64(snapdev1)) == -1) {
		warn("unable to determine volume size for %s", dev1name);
		goto out;
	}
	if ((target_volume_size = fdsize64(snapdev2)) == -1) {
		warn("unable to determine volume size for %s", dev2name);
		goto out;
	}

	for (chunk_num = 0; chunk_num < chunk_count;) {
		trace_off(printf("reading chunk "U64FMT" header\n", chunk_num););
		if ((err = fdread(deltafile, &deh, sizeof(deh))) < 0)
			goto apply_headerread_error;
		if (deh.magic_num != MAGIC_NUM)
			goto apply_magic_error;

		extent_addr = deh.extent_addr;
		extent_size = deh.num_of_chunks * chunk_size;
		if (extent_addr > target_volume_size - extent_size) /* end chunk and volume not a multiple of extent_size */
			extent_size = target_volume_size - extent_addr;
		uncomp_size = extent_size;

		if (!fullvolume && source_volume_size > extent_addr) {
			u64 source_extent_size = (extent_addr > source_volume_size - extent_size) ? (source_volume_size - extent_addr) : extent_size;
			if ((err = diskread(snapdev1, extent_data, source_extent_size, extent_addr)) < 0)
				goto apply_devread_error;
			if (posix_fadvise(snapdev1, extent_addr, source_extent_size, POSIX_FADV_DONTNEED) != 0)
				warn("can't free cached pages for the source snapshot, error %s", strerror(errno));
			/* check to see if the checksum of snap0 is the same on upstream and downstream */
			if (deh.ext1_chksum != checksum((const unsigned char *)extent_data, source_extent_size)) {
				warn("delta header checksum '%lld', actual checksum '%lld'", deh.ext1_chksum, checksum((const unsigned char *)extent_data, source_extent_size));
				goto apply_checksum_error_snap0;
			}
		}

		/* gzip compression was used on extent */
		if (deh.gzip_on == TRUE) {
			if ((err = fdread(deltafile, comp_delta, deh.extents_delta_length)) < 0)
				goto apply_deltaread_error;
			trace_off(printf("data was compressed\n"););
			/* zlib decompression */
			int comp_ret = uncompress(delta_data, (unsigned long *) &uncomp_size, comp_delta, deh.extents_delta_length);
			if (comp_ret == Z_MEM_ERROR)
				goto apply_compmem_error;
			if (comp_ret == Z_BUF_ERROR)
				goto apply_compbuf_error;
			if (comp_ret == Z_DATA_ERROR)
				goto apply_compdata_error;
		} else {
			if ((err = fdread(deltafile, delta_data, deh.extents_delta_length)) < 0)
				goto apply_deltaread_error;
			uncomp_size = deh.extents_delta_length;
		}


		//if the mode is RAW or XDELTA RAW then copy the delta file directly to updated (no uncompression)
		if (deh.mode == RAW)
			memcpy(updated, delta_data, extent_size); // !!! FIXME TODO - this has to go, bogus data copy
		if (!fullvolume && deh.mode == XDELTA) {
			trace_off(warn("read %llx chunk delta extent data starting at chunk "U64FMT"/offset "U64FMT" from \"%s\"", deh.num_of_chunks, chunk_num, extent_addr, dev1name););
			int apply_ret = apply_delta_chunk(extent_data, updated, delta_data, extent_size, uncomp_size);
			trace_off(warn("apply_ret %d\n", apply_ret););
			if (apply_ret < 0)
				goto apply_chunk_error;
		}

		if (deh.ext2_chksum != checksum((const unsigned char *)updated, extent_size))  {
			warn("deh chksum %lld, checksum %lld", deh.ext2_chksum, checksum((const unsigned char *)updated, extent_size));
			goto apply_checksum_error;
		}
		trace_off(warn("dev2name %s, extent_size %lld, extent_addr %lld", dev2name, extent_size, extent_addr););
		if ((err = diskwrite(snapdev2, updated, extent_size, extent_addr)) < 0)
			goto apply_write_error;

		if (progress_file && (((current_time = now()) - last_update) > 0)) {
			if (fsync(snapdev2))
				goto out;
			if (write_progress(progress_file, progress_tmpfile, chunk_num, chunk_count, extent_addr, tgt_snap) < 0)
				goto out;
			last_update = current_time;
		}

		chunk_num = chunk_num + deh.num_of_chunks;
	}
	trace_on(warn("All extents applied to %s\n", dev2name););
	if (fsync(snapdev2))
		goto out;
	err = progress_file ? write_progress(progress_file, progress_tmpfile, chunk_num, chunk_count, extent_addr, tgt_snap) : 0;
	goto out;

	/* error messages */
apply_headerread_error:
	warn("could not read header for extent starting at chunk "U64FMT" of "U64FMT" total chunks: %s", chunk_num, chunk_count, strerror(-err));
	goto out;

apply_magic_error:
	err = -ERANGE;
	warn("wrong magic in header for extent starting at chunk "U64FMT" of "U64FMT" total chunks", chunk_num, chunk_count);
	goto out;

apply_deltaread_error:
	warn("could not properly read delta data for extent at offset "U64FMT": %s", extent_addr, strerror(-err));
	goto out;

apply_devread_error:
	warn("could not read "U64FMT" chunk extent at offset "U64FMT" from downstream snapshot device \"%s\": %s", deh.num_of_chunks, extent_addr, dev1name, strerror(-err));
	goto out;

apply_compmem_error:
	warn("not enough buffer memory for decompression of delta for "U64FMT" chunk extent starting at offset "U64FMT, deh.num_of_chunks, extent_addr);
	goto out;

apply_compbuf_error:
	warn("not enough room in the output buffer for decompression of delta for "U64FMT" chunk extent starting at offset "U64FMT, deh.num_of_chunks, extent_addr);
	goto out;

apply_compdata_error:
	warn("compressed data corrupted in delta for "U64FMT" chunk extent starting at offset "U64FMT, deh.num_of_chunks, extent_addr);
	goto out;

apply_chunk_error:
	err = -ERANGE; /* FIXME: find better error */
	warn("delta could not be applied for "U64FMT" chunk extent with start address of "U64FMT, deh.num_of_chunks, extent_addr);
	goto out;

apply_checksum_error_snap0:
	err = -ERANGE;
	warn("checksum failed for "U64FMT" chunk extent with start address of "U64FMT" snapshot0 is not the same on the upstream and the downstream", deh.num_of_chunks, extent_addr);
	goto out;

apply_checksum_error:
	err = -ERANGE;
	warn("checksum failed for "U64FMT" chunk extent with start address of "U64FMT, deh.num_of_chunks, extent_addr);
	goto out;

apply_write_error:
	warn("updated extent could not be written at start address "U64FMT" in snapshot device \"%s\": %s", extent_addr, dev2name, strerror(-err));

out:
	if (up_extent2)
		free(up_extent2);
	if (up_extent1)
		free(up_extent1);
	if (comp_delta)
		free(comp_delta);
	if (extent_data)
		free(extent_data);
	if (delta_data)
		free(delta_data);
	if (updated)
		free(updated);
	close(snapdev2);
	if (!fullvolume)
		close(snapdev1);
	return err;
}

static int apply_delta(int deltafile, char const *devstem)
{
	struct delta_header dh;

	if (fdread(deltafile, &dh, sizeof(dh)) < 0) {
		warn("not a proper delta file (too short)");
		return -1; /* FIXME: use named error */
	}

	/* Make sure it's a proper delta file */
	if (strncmp(dh.magic, DELTA_MAGIC_ID, MAGIC_SIZE) != 0) {
		warn("not a proper delta file (wrong magic in header)");
		return -1; /* FIXME: use named error */
	}

	if (dh.chunk_size == 0) {
		warn("not a proper delta file (zero chunk size)");
		return -1; /* FIXME: use named error */
	}

	char *dev1name = NULL;
	int err, fullvolume = (dh.src_snap == ~0U);

	/* check to see if the replication is full volume replication (dh.src_snap == -1)
	 * or replication via snapshot. In case of full volume replication, pass NULL dev1name
	 * to apply_delta_extents */
	if (!fullvolume && !(dev1name = malloc_snapshot_name(devstem, dh.src_snap))) {
		warn("unable to allocate memory for dev1name");
		return -ENOMEM;
	}

	if ((err = apply_delta_extents(deltafile, dh.chunk_size, dh.chunk_num, dev1name, devstem, NULL, dh.tgt_snap)) < 0) {
		if (dev1name)
			free(dev1name);
		return err;
	}

	if (!fullvolume)
		free(dev1name);

	return 0;
}

static int ddsnap_apply_delta(char const *deltaname, char const *devstem)
{
	int deltafile;

	deltafile = open(deltaname, O_RDONLY);
	if (deltafile < 0) {
		warn("could not open delta file \"%s\" for reading", deltaname);
		return 1;
	}

	if (apply_delta(deltafile, devstem) < 0) {
		warn("could not apply delta file \"%s\" to origin device \"%s\"", deltaname, devstem);
		close(deltafile);
		return 1;
	}

	char test;

	if (read(deltafile, &test, 1) == 1)
		warn("extra data at end of delta file \"%s\"", deltaname);

	close(deltafile);

	return 0;
}

static int list_snapshots(int serv_fd, int verbose, int onlylast)
{
	int err, size;

	if ((err = outbead(serv_fd, LIST_SNAPSHOTS, struct create_snapshot, 0))) {
		warn("unable to request snaphot list: %s", strerror(-err));
		return 1;
	}

	if ((size = expect(serv_fd, SNAPSHOT_LIST)) == -1) {
		errprint("get snapshot list");
		return 1; // !!! unusual error code
	}
 
	if (size < sizeof(int32_t)) {
		warn("snapshot list reply length mismatch: expected >=%zu, actual %Lu", sizeof(int32_t), (llu_t) size);
		return 1;
	}

	int count;

	if ((err = readpipe(serv_fd, &count, sizeof(int))) < 0) {
		warn("unable to read snaphot list count: %s", strerror(-err));
		return 1;
	}

	if (size != sizeof(int32_t) + count * sizeof(struct snapinfo)) {
		warn("snapshot list reply length mismatch: expected %zu, actual %Lu", sizeof(int32_t) + count * sizeof(struct snapinfo), (llu_t)size);
		return 1;
	}


	struct snapinfo *buffer = malloc(count * sizeof(struct snapinfo));

	if (!buffer) {
		warn("unable to allocate snapshot list buffer");
		return 1;
	}

	if ((err = readpipe(serv_fd, buffer, count * sizeof(struct snapinfo))) < 0) {
		warn("unable to read snaphot list: %s", strerror(-err));
		free(buffer);
		return 1;
	}

	if (verbose)
		printf("Snapshot list:\n");

	int i;

	for (i = ((onlylast && count > 0) ? count - 1 : 0); i < count; i++) {
		if (verbose) {
			printf("Snapshot %d:\n", i);
			printf("\ttag= %u \t", buffer[i].snap);
			printf("priority= %d \t", (int)buffer[i].prio);
			printf("use count= %u \t", (unsigned int)buffer[i].usecnt);

			time_t snap_time = (time_t)buffer[i].ctime;
			char *ctime_str = ctime(&snap_time);
			if (ctime_str[strlen(ctime_str)-1] == '\n')
				ctime_str[strlen(ctime_str)-1] = '\0';

			printf("creation time= %s\n", ctime_str);
		} else {
			printf("%u ", buffer[i].snap);
		}
	}

	if (!verbose)
		printf("\n");

	free(buffer);

	return 0;
}

static int ddsnap_generate_changelist(int serv_fd, char const *changelist_filename, u32 src_snap, u32 tgt_snap)
{
	int change_fd = open(changelist_filename, O_CREAT | O_TRUNC | O_WRONLY, S_IRUSR | S_IWUSR);

	if (change_fd < 0) {
		warn("unable to open changelist file %s for writing: %s", changelist_filename, strerror(errno));
		return 1;
	}

	struct change_list *cl;

	if ((cl = stream_changelist(serv_fd, src_snap, tgt_snap)) == NULL) {
		warn("could not generate change list between snapshots %u and %u", src_snap, tgt_snap);
		return 1;
	}

	int err = write_changelist(change_fd, cl);
	free_change_list(cl);

	close(change_fd);

	if (err < 0)
		return 1;

	return 0;
}

static int delete_snapshot(int sock, u32 snaptag)
{
	int err;
	if ((err = outbead(sock, DELETE_SNAPSHOT, struct create_snapshot, snaptag)) < 0) { /* FIXME: why create_snapshot? */
		warn("unable to send delete snapshot message: %s", strerror(-err)); // !!! fixme
		return 1;
	}
	if (get_reply(sock, "delete snapshot", DELETE_SNAPSHOT_OK, 0, NULL) != 0) {
		errprint("delete snapshot");
		return 1;
	}
	return 0;
}

static int create_snapshot(int sock, u32 snaptag)
{
	int err;

	trace_on(printf("sending snapshot create request %u\n", snaptag););

	if ((err = outbead(sock, CREATE_SNAPSHOT, struct create_snapshot, snaptag)) < 0) {
		warn("unable to send create snapshot message: %s", strerror(-err)); // !!! fixme
		return 1;
	}
	if (get_reply(sock, "create snapshot", CREATE_SNAPSHOT_OK, 0, NULL) != 0) {
		errprint("create snapshot");
		return 1;
	}
	return 0;
}

static int set_priority(int sock, u32 snaptag, int8_t prio_val)
{
	int err;

	if ((err = outbead(sock, PRIORITY, struct priority_info, snaptag, prio_val)) < 0) {
		warn("unable to send set priority message: %s", strerror(-err)); // !!! fixme
		return 1;
	}
	struct priority_ok reply;
	if (get_reply(sock, "set_priority", PRIORITY_OK, sizeof(struct priority_ok), &reply) != 0) {
		errprint("set priority");
 		return 1; // !!! unusual return code
	}
	printf("%u\n", (unsigned int)reply.prio);
	return 0;
}

static int usecount(int sock, u32 snaptag, int32_t usecnt_dev)
{
	int err;

	if ((err = outbead(sock, USECOUNT, struct usecount_info, snaptag, usecnt_dev)) < 0) {
		warn("unable to send usecount message: %s", strerror(-err)); // !!! fixme
		return 1; // !!! unusual error code
	}
	struct usecount_ok reply;
	if (get_reply(sock, "usecount", USECOUNT_OK, sizeof(struct usecount_ok), &reply)) {
		errprint("set/get usecount");
		return 1;
	}
	printf("%u\n", (unsigned int)(reply.usecount));
	return 0;
}

static int ddsnap_delta_server(int lsock, char const *devstem, const char *progress_file, char const *logfile, int getsigfd)
{
	char const *origindev = devstem;
        struct pollfd pollvec[1];
	struct sigaction sigact = { .sa_handler = sighandler, .sa_flags = (SA_NOCLDSTOP | SA_NOCLDWAIT | SA_RESETHAND) };
	/* FIXME: keep list of child pids if we want to support multiple children */
	pid_t pid = -1;  /* delta server has a single child only now */

        pollvec[0] = (struct pollfd){ .fd = getsigfd, .events = POLLIN };
	if (sigprocmask(0, NULL, &sigact.sa_mask))  /* get the current signal mask */
		error("fail to set signal mask");

	for (;;) {
		int csock;

		if ((poll(pollvec, 1, 0) < 0) && (errno != EINTR))
			error("poll failed, %s", strerror(errno));
                if (pollvec[0].revents) {
                        u8 sig = 0;
                        /* it's stupid but this read also gets interrupted, so... */
                        do { } while (read(getsigfd, &sig, 1) == -1 && errno == EINTR);
                        trace_on(warn("Caught signal %i", sig););
                        switch (sig) {
                                case SIGHUP:
                                        fflush(stderr);
                                        fflush(stdout);
                                        re_open_logfile(logfile);
                                        break;
				case SIGTERM:
				case SIGINT:
					if (pid > 0)
						kill(pid, SIGKILL);
					exit(0);
				case SIGCHLD:
					pid = -1; /* delta server has a single child only now */
					break;
				default:
					break;
                        }
                }

		if ((csock = accept_socket(lsock)) < 0) {
			warn("unable to accept connection: %s", strerror(-csock));
			continue;
		}

		trace_on(fprintf(stderr, "got client connection\n"););

		assert(pid == -1); /* delta server has a single child only now */
		sigaction(SIGCHLD, &sigact, NULL); /* monitor child exits */
		if ((pid = fork()) < 0) {
			warn("unable to fork to service connection: %s", strerror(errno));
			goto cleanup_client;
		}

		if (pid != 0) {
			/* parent -- wait for another connection */
			goto cleanup_client;
		}

		trace_on(fprintf(stderr, "processing\n"););

		/* child */


		struct messagebuf message;
		int err;
		char err_msg[MAX_ERRMSG_SIZE];
		err_msg[0] = '\0';

		if ((err = readpipe(csock, &message.head, sizeof(message.head))) < 0) {
			snprintf(err_msg, MAX_ERRMSG_SIZE, "error reading upstream message header: %s", strerror(-err));
			err_msg[MAX_ERRMSG_SIZE-1] = '\0';
			goto end_connection;
		}
		if (message.head.length > maxbody) {
			snprintf(err_msg, MAX_ERRMSG_SIZE, "message body too long %u", message.head.length);
			err_msg[MAX_ERRMSG_SIZE-1] = '\0';
			goto end_connection;
		}
		if ((err = readpipe(csock, &message.body, message.head.length)) < 0) {
			snprintf(err_msg, MAX_ERRMSG_SIZE, "error reading upstream message body: %s", strerror(-err));
			err_msg[MAX_ERRMSG_SIZE-1] = '\0';
			goto end_connection;
		}

		struct delta_header body;

		switch (message.head.code) {
		case SEND_DELTA:
			if (message.head.length < sizeof(body)) {
				snprintf(err_msg, MAX_ERRMSG_SIZE, "incomplete SEND_DELTA request sent by client: length %u, size %zu", message.head.length, sizeof(body));
				err_msg[MAX_ERRMSG_SIZE-1] = '\0';
				goto end_connection;
			}

			memcpy(&body, message.body, sizeof(body));

			if (body.chunk_size == 0) {
				snprintf(err_msg, MAX_ERRMSG_SIZE, "invalid chunk size %u in SEND_DELTA", body.chunk_size);
				err_msg[MAX_ERRMSG_SIZE-1] = '\0';
				goto end_connection;
			}

			char *src_snapdev = NULL;
			if ((body.src_snap != (u32)~0UL) && !(src_snapdev = malloc_snapshot_name(devstem, body.src_snap))) {
				snprintf(err_msg, MAX_ERRMSG_SIZE, "unable to allocate device name");
				err_msg[MAX_ERRMSG_SIZE-1] = '\0';
				goto end_connection;
			}

			/* FIXME: verify snapshot exists */

			/* FIXME: In the future we should also lookup the client's address in a
			 * device permission table and check for replicatiosn already in progress.
			 */

			if (outbead(csock, SEND_DELTA_PROCEED, struct {}) < 0) {
				snprintf(err_msg, MAX_ERRMSG_SIZE, "unable to send delta proceed message to server");
				err_msg[MAX_ERRMSG_SIZE-1] = '\0';
				if (src_snapdev)
					free(src_snapdev);
				goto end_connection;
			}

			/* retrieve it */

			if (apply_delta_extents(csock, body.chunk_size,
						body.chunk_num, src_snapdev, origindev, progress_file, body.tgt_snap) < 0) {
				snprintf(err_msg, MAX_ERRMSG_SIZE, "unable to apply upstream delta to device \"%s\"", origindev);
				err_msg[MAX_ERRMSG_SIZE-1] = '\0';
				if (src_snapdev)
					free(src_snapdev);
				goto end_connection;
			}

			if (src_snapdev)
				free(src_snapdev);

			/* success */

			if (outbead(csock, SEND_DELTA_DONE, struct {}) < 0) {
				snprintf(err_msg, MAX_ERRMSG_SIZE, "unable to send delta complete message to server");
				err_msg[MAX_ERRMSG_SIZE-1] = '\0';
				goto end_connection;
			}
			trace_on(fprintf(stderr, "applied streamed delta to \"%s\", closing connection\n", origindev););
			close(csock);
			exit(0);

		default:
			snprintf(err_msg, MAX_ERRMSG_SIZE,
					"unexpected message type sent to snapshot replication server %x", message.head.code);
			err_msg[MAX_ERRMSG_SIZE-1] = '\0';
			goto end_connection;
		}

	end_connection:
		warn("closing connection on error: %s", err_msg);
		if (outhead(csock, SEND_DELTA_ERROR, strlen(err_msg)+1) < 0 ||
				writepipe(csock, err_msg, strlen(err_msg)+1) < 0)
			warn("unable to send delta error message to upstream server");
		close(csock);
		exit(1);

	cleanup_client:
		close(csock);
	}

	return 0;
}

static u32 get_state(int serv_fd, u32 snaptag)
{
	struct state_message reply;
	int err;

	if ((err = outbead(serv_fd, REQUEST_SNAPSHOT_STATE, struct status_request, snaptag))) {
		warn("unable to send state request: %s", strerror(-err)); // !!! fixme
		return 9;
	}
	if (get_reply(serv_fd, "snapshot state", SNAPSHOT_STATE, sizeof(struct state_message), &reply) != 0) {
		errprint("get snapshot state");
		return 9;
	}
	return reply.state;
}

int commas(char *buf, int len, long long n)
{
	int newlen = snprintf(buf, len, "%Lu", n);
	if (!newlen)
		return 0;
	int need = (newlen - 1) / 3;
	if ((newlen += need) >= len)
		return 0;
	char *to = buf + newlen, *from = to - need;
	*to = 0;
	while (need) {
		*--to = *--from;
		*--to = *--from;
		*--to = *--from;
		*--to = ',';
		need--;
	}
	return newlen;
}

static int ddsnap_get_status(int serv_fd, u32 snaptag, int verbose)
{
	struct status_reply *reply = generate_status(serv_fd, snaptag);
	if (!reply)
		return 1;
	int separate = !!reply->store.total;
	char number1[27], number2[27];
	unsigned snapshots = reply->snapshots;

	commas(number1, sizeof(number1), reply->meta.total);
	commas(number2, sizeof(number2), reply->meta.free);
	printf(separate? "Snapshot metadata " : "Snapshot store ");
	printf("block size = %lu; ", 1UL << reply->meta.chunksize_bits);
	printf("%s of ", number2);
	printf("%s chunks free\n", number1);

	if (separate) {
		commas(number1, sizeof(number1), reply->store.total);
		commas(number2, sizeof(number2), reply->store.free);
		printf("Snapshot data store ");
		printf("chunk size = %lu; ", 1UL << reply->store.chunksize_bits);
		printf("%s of ", number2);
		printf("%s chunks free\n", number1);
	}

	commas(number1, sizeof(number1), get_snapshot_sectors(serv_fd, (u32)~0UL) << SECTOR_BITS); // include this in status reply!!
	printf("Origin size: %s bytes\n", number1);
	printf("Write density: %g\n", (double)reply->write_density/(double)0xffffffff);

	time_t snaptime = (time_t)reply->ctime;
	char *ctime_str = ctime(&snaptime);
	if (ctime_str[strlen(ctime_str)-1] == '\n')
		ctime_str[strlen(ctime_str)-1] = '\0';
	printf("Creation time: %s\n", ctime_str);

	unsigned int row, col;
	u64 total_chunks;

	printf("%6s %24s %6s %4s %8s %8s", "Snap", "Creation time", "Usecnt", "Prio", "Chunks", "Unshared");

	if (verbose) {
		for (col = 1; col < snapshots; col++)
			printf(" %7dX", col);
	} else {
		printf(" %8s", "Shared");
	}

	printf("\n");

	for (row = 0; row < snapshots; row++) {
		struct snapshot_details *details = snapshot_details(reply, row, snapshots);

		if (snaptag != (u32)~0UL && details->snapinfo.snap != snaptag)
			continue;

		printf("%6u", details->snapinfo.snap);


		snaptime = (time_t)details->snapinfo.ctime;
		ctime_str = ctime(&snaptime);
		if (ctime_str[strlen(ctime_str)-1] == '\n')
			ctime_str[strlen(ctime_str)-1] = '\0';
		printf(" %24s %6u %4i", ctime_str, details->snapinfo.usecnt, details->snapinfo.prio);

		if (details->sharing[0] == -1) {
			printf(" %8s %8s %8s\n", "!", "!", "!");
			continue;
		}

		total_chunks = 0;
		for (col = 0; col < snapshots; col++)
			total_chunks += details->sharing[col];

		printf(" %8llu", (llu_t) total_chunks);
		printf(" %8llu", (llu_t) details->sharing[0]);

		if (verbose)
			for (col = 1; col < snapshots; col++)
				printf(" %8llu", (llu_t) details->sharing[col]);
		else
			printf(" %8llu", (llu_t)(total_chunks - details->sharing[0]));

		printf("\n");
	}

	/* if we are printing all the snapshot stats, also print totals for each
	 * degree of sharing
	 */

	if (snaptag == (u32)~0UL) {
		u64 *column_totals;

		if (!(column_totals = malloc(sizeof(u64) * snapshots))) {
			warn("unable to allocate array for column totals");
			free(reply);
			return 1;
		}

		/* sum the columns and divide by their share counts */

		total_chunks = 0;
		for (col = 0; col < snapshots; col++) {
			column_totals[col] = 0;
			for (row = 0; row < snapshots; row++) {
				struct snapshot_details *details = snapshot_details(reply, row, snapshots);
				if (details->sharing[0] == -1)
					continue;
				column_totals[col] += details->sharing[col];
			}
			column_totals[col] /= col+1;

			total_chunks += column_totals[col];
		}

		printf("%6s %45llu", "totals", total_chunks);
		if (snapshots > 0)
			printf(" %8llu", column_totals[0]);
		else
			printf(" %8d", 0);

		if (verbose)
			for (col = 1; col < snapshots; col++)
				printf(" %8llu", column_totals[col]);
		else if (snapshots > 0)
			printf(" %8llu", total_chunks - column_totals[0]);
		else
			printf(" %8d", 0);

		printf("\n");

		free(column_totals);

		if (separate)
			commas(number1, sizeof(number1), (reply->meta.total - reply->meta.free) << reply->meta.chunksize_bits);
		else
			commas(number1, sizeof(number1), (reply->meta.total - reply->meta.free - total_chunks) << reply->meta.chunksize_bits);
		printf("Metadata usage: %s bytes\n", number1);
	}

	free(reply);

	return 0;
}

static int ddsnap_resize_devices(int serv_fd, u64 orgsize, u64 snapsize, u64 metasize)
{
	int err;
	if ((err = outbead(serv_fd, RESIZE, struct resize_request, orgsize, snapsize, metasize))) {
		warn("unable to send resize request: %s", strerror(-err));
		return -1;
	}
	struct resize_request reply;
	if (get_reply(serv_fd, "resize", RESIZE, sizeof(struct resize_request), &reply) != 0) {
		errprint("resize device");
		return -1;
	}
	printf("Device sizes after resizing:\n\t");
	printf("origin device %Lu\n\t", (unsigned long long) reply.orgsize);
	printf("snapshot device %Lu\n\t", (unsigned long long) reply.snapsize);
	printf("metadata device %Lu\n", (unsigned long long) reply.metasize);

	if ((orgsize && reply.orgsize != orgsize) || (snapsize && reply.snapsize != snapsize) || (metasize && reply.metasize != metasize)) {
		warn("Can not resize devices to the specified value: orgsize %Lu snapsize %Lu metasize %Lu!!", orgsize, snapsize, metasize);
		return -1;
	}

	return 0;
}

static void mainUsage(void)
{
	printf("usage: ddsnap [-?|--help|--usage|--version] <subcommand>\n"
	       "\n"
	       "Available subcommands:\n"
	       "        initialize        Initialize snapshot storage device\n"
	       "        agent             Start the snapshot agent\n"
	       "        server            Start the snapshot server\n"
	       "	create            Create a snapshot\n"
	       "	delete            Delete a snapshot\n"
	       "	list              Return list of snapshots currently held\n"
	       "	priority          Set the priority of a snapshot\n"
	       "	usecount          Change the use count of a snapshot\n"
	       "        status            Report snapshot usage statistics\n"
	       "	vol               \n"
               "        usage: ddsnap vol [-?|--help|--usage] <subcommand>\n"
               "\n"
               "        Available vol subcommands:\n"
	       "	create            Create a volume delta file\n"
	       "	apply             Apply a volume file to a downstream volume\n"
	       "	send              Send a volume file to a downstream server\n"
	       "        listen            Listen for a volume file arriving from upstream\n"
	       "	delta             \n"
               "        usage: ddsnap delta [-?|--help|--usage] <subcommand>\n"
               "\n"
               "        Available delta subcommands:\n"
               "                changelist        Create a changelist given 2 snapshots\n"
	       "	        create            Create a delta file given a changelist and 2 snapshots\n"
	       "	        apply             Apply a delta file to a volume\n"
	       "	        send              Send a delta file to a downstream server\n"
	       "	        listen            Listen for a delta arriving from upstream\n");
}

static void deltaUsage(void)
{
	printf("usage: ddsnap delta [-?|--help|--usage] <subcommand>\n"
	       "\n"
               "Available delta subcommands:\n"
               "        changelist        Create a changelist given 2 snapshots\n"
	       "	create            Create a delta file given a changelist and 2 snapshots\n"
	       "	apply             Apply a delta file to a volume\n"
	       "	send              Send a delta file to a downstream server\n"
	       "        listen            Listen for a delta arriving from upstream\n");
}

static void cdUsage(poptContext optCon, int exitcode, char const *error, char const *addl)
{
	poptPrintUsage(optCon, stderr, 0);
	if (error) fprintf(stderr, "%s: %s", error, addl);
	exit(exitcode);
}

#ifdef DDSNAP_MEM_MONITOR
int mmon_interval = DDSNAP_MEM_MONITOR;
#endif

#define DEVMAP_PATH "/dev/mapper"
int main(int argc, char *argv[])
{
	char const *command;

	struct poptOption noOptions[] = {
		POPT_TABLEEND
	};

	char *js_str = NULL, *bs_str = NULL, *cs_str = NULL;
	int yes = FALSE;
	struct poptOption initOptions[] = {
		{ "yes", 'y', POPT_ARG_NONE, &yes, 0, "Answer yes to all prompts", NULL},
		{ "journalsize", 'j', POPT_ARG_STRING, &js_str, 0, "User specified journal size, i.e. 400k (default: 100 * chunk_size)", "size" },
		{ "blocksize", 'b', POPT_ARG_STRING, &bs_str, 0, "Snapshot metadata block size (power of two, default = 4K)", "size" },
		{ "chunksize", 'c', POPT_ARG_STRING, &cs_str, 0, "Snapshot store chunk size (power of two, default = 16K)", "size" },
		POPT_TABLEEND
	};

	int debug = 0, experimental = 0, nobg = 0;
	char const *logfile = NULL;
	char const *pidfile = NULL;
	char const *progress_file = NULL;
	char const *resume = NULL;
	char const *cachesize_str = NULL;
	char const *ratelimit_str = NULL;
	struct poptOption serverOptions[] = {
		{ "debug", 'D', POPT_ARG_NONE, &debug, 0, "turn on debugging checks", NULL }, // !!! should turn on debug logging too
		{ "experimental", 'X', POPT_ARG_NONE, &experimental, 0, "use experimental optimizations", NULL },
		{ "foreground", 'f', POPT_ARG_NONE, &nobg, 0, "run in foreground. daemonized by default.", NULL }, // !!! unusual semantics, we should be foreground by default, and optionally daemonize
		{ "logfile", 'l', POPT_ARG_STRING, &logfile, 0, "use specified log file", NULL },
		{ "cachesize", 'k', POPT_ARG_STRING, &cachesize_str, 0, "Buffer cache size (default = max(128M,1/4 sys RAM)", "size" },
#ifdef DDSNAP_MEM_MONITOR
		{ "mmonitor", 'm', POPT_ARG_INT, &mmon_interval, 0, "Memory monitor delay, seconds, zero to disable.", NULL },
#endif
		{ "pidfile", 'p', POPT_ARG_STRING, &pidfile, 0, "use specified process id file", NULL },
		{ "progress", 'o', POPT_ARG_STRING, &progress_file, 0, "Output progress to specified file", NULL },
		POPT_TABLEEND
	};


	int xd = FALSE, raw = FALSE, best_comp = FALSE, gzip_level = DEF_GZIP_COMP;
	struct poptOption cdOptions[] = {
		{ "xdelta", 'x', POPT_ARG_NONE, &xd, 0, "Delta file format: xdelta chunk", NULL },
		{ "raw", 'r', POPT_ARG_NONE, &raw, 0, "Delta file format: raw chunk from later snapshot", NULL },
		{ "best", 'b', POPT_ARG_NONE, &best_comp, 0, "Delta file format: best compression (slowest)", NULL},
		{ "gzip", 'g', POPT_ARG_INT, &gzip_level, 0, "Compression via gzip", "compression_level"},
		{ "progress", 'p', POPT_ARG_STRING, &progress_file, 0, "Output progress to specified file", NULL },
		{ "resume", 's', POPT_ARG_STRING, &resume, 0, "Resume from specified address", NULL },
		{ "ratelimit", 'l', POPT_ARG_STRING, &ratelimit_str, 0, "Rate limit to send delta to downstream (unit = bytes/s; default = 0, no limit)", "rate" },
		POPT_TABLEEND
	};

	int last = FALSE;
	int list = FALSE;
	int size = FALSE;
	int state = FALSE;
	int verb = FALSE;
	struct poptOption stOptions[] = {
		{ "last", '\0', POPT_ARG_NONE, &last, 0, "List the newest snapshot", NULL},
		{ "list", 'l', POPT_ARG_NONE, &list, 0, "List all active snapshots", NULL},
		{ "size", 's', POPT_ARG_NONE, &size, 0, "Print the size of the origin device", NULL},
		{ "state", 'S', POPT_ARG_NONE, &state, 0,
			"Return the state of snapshot after exit. Output: 0: normal; 1: not exist; 2: squashed; 9: other error", NULL},
		{ "verbose", 'v', POPT_ARG_NONE, &verb, 0, "Verbose sharing information", NULL},
		POPT_TABLEEND
	};

	struct poptOption deltaOptions[] = {
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &noOptions, 0,
		  "Create changelist\n\t Function: Create a changelist given 2 snapshots\n\t Usage: delta changelist <sockname> <changelist> <snapshot1> <snapshot2>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &cdOptions, 0,
		  "Create delta\n\t Function: Create a delta file given a changelist and 2 snapshots\n\t Usage: delta create [OPTION...] <changelist> <deltafile> <devstem>\n", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &noOptions, 0,
		  "Apply delta\n\t Function: Apply a delta file to a volume\n\t Usage: delta apply <deltafile> <devstem>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &serverOptions, 0,
		  "Listen\n\t Function: Listen for a delta arriving from upstream\n\t Usage: delta listen [OPTION...] <devstem> [<host>[:<port>]]", NULL },
		POPT_TABLEEND
	};

	char *orgsize_str = NULL, *snapsize_str = NULL, *metasize_str = NULL;
	struct poptOption resizeOptions[] = {
		{ "origin", 'o', POPT_ARG_STRING, &orgsize_str, 0, "New origin device size", "size" },
		{ "snapshot", 's', POPT_ARG_STRING, &snapsize_str, 0, "New snapshot device size", "size" },
		{ "metadata", 'm', POPT_ARG_STRING, &metasize_str, 0, "New metadata device size", "size" },
		POPT_TABLEEND
	};

	poptContext mainCon;
	struct poptOption mainOptions[] = {
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &initOptions, 0,
		  "Initialize\n\t Function: Initialize a snapshot storage volume\n\t Usage: initialize [OPTION...] <dev/snapshot> <dev/origin> [dev/meta]", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &serverOptions, 0,
		  "Agent Server\n\t Function: Start the snapshot agent\n\t Usage: agent [OPTION...] <agent_socket>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &serverOptions, 0,
		  "Snapshot Server\n\t Function: Start the snapshot server\n\t Usage: server [OPTION...] <dev/snapshot> <dev/origin> [dev/meta] <agent_socket> <server_socket>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &noOptions, 0,
		  "Create snapshot\n\t Function: Create a snapshot\n\t Usage: create <sockname> <snapshot>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &noOptions, 0,
		  "Delete snapshot\n\t Function: Delete a snapshot\n\t Usage: delete <sockname> <snapshot>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &noOptions, 0,
		  "List snapshots\n\t Function: Return list of snapshots currently held\n\t Usage: list <sockname>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &noOptions, 0,
		  "Priority\n\t Function: Set the priority of a snapshot\n\t Usage: priority <sockname> <snap_tag> <new_priority_value>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &noOptions, 0,
		  "Usecount\n\t Function: Change the use count of a snapshot\n\t Usage: usecount <sockname> <snap_tag> <diff_amount>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &stOptions, 0,
		  "Get statistics\n\t Function: Report snapshot usage statistics\n\t Usage: status [OPTION...] <sockname> [<snapshot>]", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &deltaOptions, 0,
		  "Delta\n\t Usage: delta [OPTION...] <subcommand> ", NULL},
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &cdOptions, 0,
		  "Transmit\n\t Function: Stream delta of the given two snapshots to downstream\n\t Usage: transmit [OPTION...] <sockname> <host>[:<port>] [snapshot1] <snapshot2>", NULL},
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &resizeOptions, 0,
		  "Resize\n\t Function: Change origin/snapshot/metadata device size\n\t Usage: resize [OPTION...] <sockname>", NULL },
		{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &noOptions, 0,
		  "Revert to snapshot\n\t Function: Revert to a snapshot\n\t Usage: revert <sockname> <snapshot>", NULL },
		{ "version", 'V', POPT_ARG_NONE, NULL, 0, "Show version", NULL },
		POPT_AUTOHELP
		POPT_TABLEEND
	};
	mainCon = poptGetContext(NULL, argc, (const char **)argv, mainOptions, 0);

	if (argc < 2) {
		poptPrintHelp(mainCon, stdout, 0);
		poptFreeContext(mainCon);
		exit(1);
	}

	command = argv[1];

	if (strcmp(command, "--help") == 0 || strcmp(command, "-?") == 0) {
		poptPrintHelp(mainCon, stdout, 0);
		poptFreeContext(mainCon);
		exit(0);
	}

	poptFreeContext(mainCon);

	if (strcmp(command, "--version") == 0 || strcmp(command, "-V") == 0) {
		printf("ddsnap %s built on %s by %s@%s\n",
			VERSION_STRING, BUILD_DATE, BUILD_USER, BUILD_HOST);
		exit(0);
	}

	if (strcmp(command, "--usage") == 0) {
		mainUsage();
		exit(0);
	}

	char process_name[16];
	snprintf(process_name, 16, "ddsnap_%s", command);
	prctl(PR_SET_NAME, process_name);
	if (!strcmp(command, "dump")) {
		int err = 1;
		u64 start = 0, finish = -1; /* if not specified, the defaults dump the entire tree */
		if (argc == 5) {
			char *endptr;
			start = strtoull(argv[3], &endptr, 10);
			if (*endptr == '\0')
				finish = strtoull(argv[4], &endptr, 10);
			if (*endptr != '\0')
				fprintf(stderr, "%s %s: invalid chunk specified\n", argv[0], argv[1]);
			else
				err = 0;
		}
		if ((argc != 3) && err) {
			printf("Usage: %s dump <socket> [<start> <end>]\n", argv[0]);
			exit(1);
		}
		int sock = create_socket(argv[2]);
		if ((err = outbead(sock, DUMP_TREE_RANGE, struct dump_tree_range, start, finish))) {
			warn("unable to send dump tree request: %s", strerror(-err));
			exit(1);
		}
		exit(0);
	}

	if (strcmp(command, "initialize") == 0) {
		char const *snapdev, *origdev, *metadev;
		u32 bs_bits = DEFAULT_CHUNK_SIZE_BITS;
		u32 cs_bits = DEFAULT_CHUNK_SIZE_BITS;
		u32 js_bytes = DEFAULT_JOURNAL_SIZE;

		struct poptOption options[] = {
			{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &initOptions, 0, NULL, NULL },
			POPT_AUTOHELP
			POPT_TABLEEND
		};

		poptContext initCon = poptGetContext(NULL, argc-1, (const char **)&(argv[1]), options, 0);
		poptSetOtherOptionHelp(initCon, "<dev/snapshot> <dev/origin> [dev/meta]");

		char initOpt = poptGetNextOpt(initCon);

		if (argc < 3) {
			poptPrintUsage(initCon, stderr, 0);
			exit(1);
		}

		if (initOpt < -1) {
			fprintf(stderr, "%s: %s: %s\n", command, poptBadOption(initCon, POPT_BADOPTION_NOALIAS), poptStrerror(initOpt));
			poptFreeContext(initCon);
			return 1;
		}

		snapdev = poptGetArg(initCon);
		if (!snapdev) {
			fprintf(stderr, "%s: snapshot device must be specified\n", command);
			poptPrintUsage(initCon, stderr, 0);
			poptFreeContext(initCon);
			return 1;
		}

		origdev = poptGetArg(initCon);
		if (!origdev) {
			fprintf(stderr, "%s: origin device must be specified\n", command);
			poptPrintUsage(initCon, stderr, 0);
			poptFreeContext(initCon);
			return 1;
		}

		metadev = poptGetArg(initCon); /* ok if NULL */

		trace_off(printf("snapdev is %s, origdev is %s, metadev is %s.\n", snapdev, origdev, metadev););

		if (is_same_device(snapdev, origdev)) {
			poptPrintUsage(initCon, stderr, 0);
			poptFreeContext(initCon);
			return 1;
		}

		if (bs_str && cs_str && (strcmp(bs_str, cs_str) != 0)) {

			if (!metadev) {
				fprintf(stderr, "%s: metadata device must be specified\n", command);
				poptPrintUsage(initCon, stderr, 0);
				poptFreeContext(initCon);
				return 1;
			}

			if (is_same_device(snapdev, metadev)) {
				poptPrintUsage(initCon, stderr, 0);
				poptFreeContext(initCon);
				return 1;
			}


			if (is_same_device(origdev, metadev)) {
				poptPrintUsage(initCon, stderr, 0);
				poptFreeContext(initCon);
				return 1;
			}
		}

		if (poptPeekArg(initCon) != NULL) {
			fprintf(stderr, "%s: too many arguments\n", command);
			poptPrintUsage(initCon, stderr, 0);
			poptFreeContext(initCon);
			return 1;
		}

		int orgdev_, snapdev_, metadev_;
		if ((snapdev_ = open(snapdev, O_RDWR | O_DIRECT)) == -1)
			error("Could not open snapshot store %s: %s", snapdev, strerror(errno));

		if ((orgdev_ = open(origdev, O_RDONLY | O_DIRECT)) == -1)
			error("Could not open origin volume %s: %s", origdev, strerror(errno));

		metadev_ = snapdev_;

		if (metadev && !is_same_device(snapdev, metadev) && (metadev_ = open(metadev, O_RDWR | O_DIRECT)) == -1)
			error("Could not open meta volume %s: %s", metadev, strerror(errno));

		if (!yes && sniff_snapstore(metadev_) > 0) {
			printf("There exists a valid magic number in sb, are you sure you want to overwrite? (y/N) ");
			if (toupper(getchar()) != 'Y')
				return 1;
		}

		poptFreeContext(initCon);

		trace_off(warn("js_bytes was %u, bs_bits was %u and cs_bits was %u", js_bytes, bs_bits, cs_bits););

		if (bs_str != NULL) {
			bs_bits = strtobits(bs_str);
			if (bs_bits == INPUT_ERROR || bs_bits < SECTOR_BITS) {
				poptPrintUsage(initCon, stderr, 0);
				fprintf(stderr, "Invalid block size input. Try 64k\n");
				exit(1);
			}
			if (!cs_str)
				cs_bits = bs_bits;
		}

		if (cs_str != NULL) {
			cs_bits = strtobits(cs_str);
			if (cs_bits == INPUT_ERROR || cs_bits < SECTOR_BITS) {
				poptPrintUsage(initCon, stderr, 0);
				fprintf(stderr, "Invalid chunk size input. Try 64k\n");
				exit(1);
			}
			if (!bs_str)
				bs_bits = cs_bits;
		}

		if (js_str != NULL) {
			js_bytes = strtobytes(js_str);
			if (js_bytes == INPUT_ERROR || js_bytes < SECTOR_BITS) {
				poptPrintUsage(initCon, stderr, 0);
				fprintf(stderr, "Invalid journal size input. Try 400k\n");
				exit(1);
			}
		}
		trace_off(printf("js_bytes is %u, bs_bits is %u, and cs_bits is %u\n", js_bytes, bs_bits, cs_bits););
		return init_snapstore(orgdev_, snapdev_, metadev_, bs_bits, cs_bits, js_bytes);
	}
	if (strcmp(command, "agent") == 0) {
		char const *sockname;
		int listenfd;

		struct poptOption options[] = {
			{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &serverOptions, 0, NULL, NULL },
			POPT_AUTOHELP
			POPT_TABLEEND
		};

		poptContext agentCon = poptGetContext(NULL, argc-1, (const char **)&(argv[1]), options, 0);
		poptSetOtherOptionHelp(agentCon, "<agent_socket>");

		char agentOpt = poptGetNextOpt(agentCon);

		if (agentOpt < -1) {
			fprintf(stderr, "%s: %s: %s\n", command, poptBadOption(agentCon, POPT_BADOPTION_NOALIAS), poptStrerror(agentOpt));
			poptFreeContext(agentCon);
			return 1;
		}

		sockname = poptGetArg(agentCon);
		if (sockname == NULL) {
			fprintf(stderr, "%s: socket name for ddsnap agent must be specified\n", command);
			poptPrintUsage(agentCon, stderr, 0);
			poptFreeContext(agentCon);
			return 1;
		}
		if (poptPeekArg(agentCon) != NULL) {
			fprintf(stderr, "%s: only one socket name may be specified\n", command);
			poptPrintUsage(agentCon, stderr, 0);
			poptFreeContext(agentCon);
			return 1;
		}

		poptFreeContext(agentCon);

		if (monitor_setup(sockname, &listenfd) < 0)
			error("Could not setup ddsnap agent server\n");

		int getsigfd;
		if (!nobg) {
			pid_t pid;

			if (!logfile)
				logfile = "/var/log/ddsnap.agent.log";

			pid = daemonize(logfile, pidfile, &getsigfd);
			if (pid == -1)
				error("Could not daemonize\n");
			if (pid != 0) {
				trace_off(printf("pid = %lu\n", (unsigned long)pid););
				return 0;
			}
		}

		if (monitor(listenfd, &(struct context){ .polldelay = -1 },
		    logfile, getsigfd) < 0)
			error("Could not start ddsnap agent server\n");

		return 0; /* not reached */

	}
	if (strcmp(command, "server") == 0) {
		char const *snapdev, *origdev, *metadev;
		char const *agent_sockname, *server_sockname;
		struct poptOption options[] = {
			{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &serverOptions, 0, NULL, NULL },
			POPT_AUTOHELP
			POPT_TABLEEND
		};

		poptContext serverCon = poptGetContext(NULL, argc-1, (const char **)&(argv[1]), options, 0);
		poptSetOtherOptionHelp(serverCon, "<dev/snapshot> <dev/origin> [dev/meta] <agent_socket> <server_socket>");

		char serverOpt = poptGetNextOpt(serverCon);

		if (argc < 3) {
			poptPrintUsage(serverCon, stderr, 0);
			exit(1);
		}

		if (serverOpt < -1) {
			fprintf(stderr, "%s: %s: %s\n", command, poptBadOption(serverCon, POPT_BADOPTION_NOALIAS), poptStrerror(serverOpt));
			poptFreeContext(serverCon);
			return 1;
		}

		unsigned long long cachesize_bytes = 0;
		if (cachesize_str != NULL) {
			cachesize_bytes = strtobytes64(cachesize_str);
			if (cachesize_bytes == INPUT_ERROR_64) {
				fprintf(stderr, "Invalid cache size input. Omit option, or use 0 for the default\n");
				poptPrintUsage(serverCon, stderr, 0);
				return 1;
			}
		}

		snapdev = poptGetArg(serverCon);
		if (!snapdev) {
			fprintf(stderr, "%s: snapshot device must be specified\n", command);
			poptPrintUsage(serverCon, stderr, 0);
			poptFreeContext(serverCon);
			return 1;
		}

		origdev = poptGetArg(serverCon);
		if (!origdev) {
			fprintf(stderr, "%s: origin device must be specified\n", command);
			poptPrintUsage(serverCon, stderr, 0);
			poptFreeContext(serverCon);
			return 1;
		}

		char const *extra_arg[3];

		extra_arg[0] = poptGetArg(serverCon);
		extra_arg[1] = poptGetArg(serverCon);
		extra_arg[2] = poptGetArg(serverCon);

		if (!extra_arg[0] || !extra_arg[1]) {
			fprintf(stderr, "%s: agent and server socket names must both be specified\n", command);
			poptPrintUsage(serverCon, stderr, 0);
			poptFreeContext(serverCon);
			return 1;
		}

		if (extra_arg[2]) {
			metadev = extra_arg[0];
			agent_sockname = extra_arg[1];
			server_sockname = extra_arg[2];
		} else {
			metadev = NULL;
			agent_sockname = extra_arg[0];
			server_sockname = extra_arg[1];
		}


		if (poptPeekArg(serverCon) != NULL) {
			fprintf(stderr, "%s: too many arguments\n", command);
			poptPrintUsage(serverCon, stderr, 0);
			poptFreeContext(serverCon);
			return 1;
		}

		int orgdev_, snapdev_, metadev_;
		if ((snapdev_ = open(snapdev, O_RDWR | O_DIRECT)) == -1)
			error("Could not open snapshot store %s: %s", snapdev, strerror(errno));

		if ((orgdev_ = open(origdev, O_RDONLY | O_DIRECT)) == -1)
			error("Could not open origin volume %s: %s", origdev, strerror(errno));

		metadev_ = snapdev_;
		if (metadev && !is_same_device(snapdev, metadev) && (metadev_ = open(metadev, O_RDWR | O_DIRECT)) == -1)
			error("Could not open meta volume %s: %s", metadev, strerror(errno));

		poptFreeContext(serverCon);

		enum runflags flags = experimental * RUN_DEFER | debug * RUN_SELFCHECK;

		return start_server(
			orgdev_, snapdev_, metadev_,
			agent_sockname, server_sockname, logfile, pidfile,
			nobg, cachesize_bytes, flags);
	}
	if (strcmp(command, "create") == 0) {
		if (argc != 4) {
			printf("Usage: %s create <sockname> <snapshot>\n", argv[0]);
			return 1;
		}

		u32 snaptag;

		if (parse_snaptag(argv[3], &snaptag) < 0) {
			fprintf(stderr, "%s: invalid snapshot %s\n", argv[0], argv[3]);
			return 1;
		}

		int sock = create_socket(argv[2]);

		int ret = create_snapshot(sock, snaptag);
		close(sock);
		return ret;
	}
	if (strcmp(command, "delete") == 0) {
		if (argc != 4) {
			printf("Usage: %s delete <sockname> <snapshot>\n", argv[0]);
			return 1;
		}

		u32 snaptag;

		if (parse_snaptag(argv[3], &snaptag) < 0) {
			fprintf(stderr, "%s: invalid snapshot %s\n", argv[0], argv[3]);
			return 1;
		}

		int sock = create_socket(argv[2]);

		int ret = delete_snapshot(sock, snaptag);
		close(sock);
		return ret;
	}
	if (strcmp(command, "list") == 0) {
		if (argc != 3) {
			printf("Usage: %s list <sockname>\n", argv[0]);
			return 1;
		}

		int sock = create_socket(argv[2]);

		int ret = list_snapshots(sock, TRUE, FALSE);
		close(sock);
		return ret;
	}
	if (strcmp(command, "priority") == 0) {
		if (argc != 5) {
			printf("usage: %s priority <sockname> <snap_tag> <new_priority_value>\n", argv[0]);
			return 1;
		}

		u32 snaptag;

		if (parse_snaptag(argv[3], &snaptag) < 0) {
			fprintf(stderr, "%s: invalid snapshot %s\n", argv[0], argv[3]);
			return 1;
		}

		int sock = create_socket(argv[2]);

		int prio = atoi(argv[4]);
		if (prio < -128 || prio > 127) {
			fprintf(stderr, "%s: priority %d out of range\n", argv[0], prio);
			return 1;
		}
		int ret = set_priority(sock, snaptag, prio);
		close(sock);
		return ret;
	}
	if (strcmp(command, "usecount") == 0) {
		int diff_amount = 0;

		if (argc != 4 && argc != 5) {
			printf("usage: %s usecount <sockname> <snap_tag> [diff_amount]\n", argv[0]);
			return 1;
		}
		if (argc == 5)
			diff_amount = atoi(argv[4]);

		u32 snaptag;

		if (parse_snaptag(argv[3], &snaptag) < 0) {
			fprintf(stderr, "%s: invalid snapshot %s\n", argv[0], argv[3]);
			return 1;
		}

		int sock = create_socket(argv[2]);

		int ret = usecount(sock, snaptag, diff_amount);
		close(sock);
		return ret;
	}
	if (strcmp(command, "status") == 0) {
		poptContext cdCon;

		struct poptOption options[] = {
			{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &stOptions, 0, NULL, NULL },
			POPT_AUTOHELP
			POPT_TABLEEND
		};

		cdCon = poptGetContext(NULL, argc-1, (const char **)&(argv[1]), options, 0);
		poptSetOtherOptionHelp(cdCon, "<sockname> [<snapshot>]");

		char c;

		while ((c = poptGetNextOpt(cdCon)) >= 0);
		if (c < -1) {
			fprintf(stderr, "%s: %s: %s\n", argv[0], poptBadOption(cdCon, POPT_BADOPTION_NOALIAS), poptStrerror(c));
			poptFreeContext(cdCon);
			return 1;
		}

		if (last+list+size+state+verb > 1) {
			fprintf(stderr, "%s %s: Incompatible status options specified\n", argv[0], argv[1]);
			poptPrintUsage(cdCon, stderr, 0);
			poptFreeContext(cdCon);
			return 1;
		}

		if (last) {
			char const *sockname;

			sockname = poptGetArg(cdCon);

			if (sockname == NULL)
				cdUsage(cdCon, 1, argv[0], "Must specify socket name to status\n");
			if (poptPeekArg(cdCon) != NULL)
				cdUsage(cdCon, 1, argv[0], "Too many arguments to status\n");

			poptFreeContext(cdCon);

			int sock = create_socket(sockname);

			int ret = list_snapshots(sock, FALSE, TRUE);
			close(sock);

			return ret;
		} else if (list) {
			char const *sockname;

			sockname = poptGetArg(cdCon);

			if (sockname == NULL)
				cdUsage(cdCon, 1, argv[0], "Must specify socket name to status\n");
			if (poptPeekArg(cdCon) != NULL)
				cdUsage(cdCon, 1, argv[0], "Too many arguments to status\n");

			poptFreeContext(cdCon);

			int sock = create_socket(sockname);

			int ret = list_snapshots(sock, FALSE, FALSE);
			close(sock);

			return ret;
		} else if (size) {
			char const *sockname, *snaptagstr;

			sockname = poptGetArg(cdCon);
			snaptagstr = poptGetArg(cdCon);

			if (sockname == NULL)
				cdUsage(cdCon, 1, argv[0], "Must specify socket name to status\n");
			if (poptPeekArg(cdCon) != NULL)
				cdUsage(cdCon, 1, argv[0], "Too many arguments to status\n");

			poptFreeContext(cdCon);


			int sock = create_socket(sockname);

			u32 snaptag;
			if (snaptagstr == NULL)
				snaptag = ~((u32)0U); /* meaning "origin snapshot" */
			else if (parse_snaptag(snaptagstr, &snaptag) < 0) {
				fprintf(stderr, "%s: invalid snapshot %s\n", argv[0], snaptagstr);
				return 1;
			}
			printf("%Lu\n", get_snapshot_sectors(sock, snaptag));
			close(sock);
			return 0;
		} else if (state) {
			char const *sockname, *snaptagstr;

			sockname = poptGetArg(cdCon);
			snaptagstr = poptGetArg(cdCon);

			if (sockname == NULL)
				cdUsage(cdCon, 9, argv[0], "Must specify socket name to state\n");
			if (snaptagstr == NULL)
				cdUsage(cdCon, 9, argv[0], "Must specify snapshot id to state\n");
			if (poptPeekArg(cdCon) != NULL)
				cdUsage(cdCon, 9, argv[0], "Too many arguments to state\n");

			poptFreeContext(cdCon);

			u32 snaptag;

			if (parse_snaptag(snaptagstr, &snaptag) < 0) {
				fprintf(stderr, "%s: invalid snapshot %s\n", argv[0], snaptagstr);
				return 1;
			}

			int sock = create_socket(sockname);

			int ret = get_state(sock, snaptag);
			close(sock);

			return ret;
		} else{
			char const *sockname, *snaptagstr;

			sockname   = poptGetArg(cdCon);
			snaptagstr = poptGetArg(cdCon);

			if (sockname == NULL)
				cdUsage(cdCon, 1, argv[0], "Must specify socket name to status\n");
			if (poptPeekArg(cdCon) != NULL)
				cdUsage(cdCon, 1, argv[0], "Too many arguments to status\n");

			poptFreeContext(cdCon);

			u32 snaptag;

			if (snaptagstr) {
				if (parse_snaptag(snaptagstr, &snaptag) < 0) {
					fprintf(stderr, "%s: invalid snapshot %s\n", argv[0], snaptagstr);
					return 1;
				}
			} else {
				snaptag = ~((u32)0U); /* meaning "all snapshots" */
			}

			int sock = create_socket(sockname);

			int ret = ddsnap_get_status(sock, snaptag, verb);
			close(sock);

			return ret;
		}
	}
	/* syntax for delta/volume replication: ddsnap transmit <socket> <host[:port]> [fromsnap] <tosnap> */
	if (strcmp(command, "transmit") == 0) {
		poptContext cdCon;

		struct poptOption options[] = {
			{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &cdOptions, 0, NULL, NULL },
			POPT_AUTOHELP
			POPT_TABLEEND
		};

		cdCon = poptGetContext(NULL, argc-1, (const char **)&(argv[1]), options, 0);
		poptSetOtherOptionHelp(cdCon, "<sockname> <host[:port]> [fromsnap] <tosnap>");

		char c;

		while ((c = poptGetNextOpt(cdCon)) >= 0);
		if (c < -1) {
			fprintf(stderr, "%s: %s: %s\n", argv[0], poptBadOption(cdCon, POPT_BADOPTION_NOALIAS), poptStrerror(c));
			poptFreeContext(cdCon);
			return 1;
		}

		/* Make sure the options are mutually exclusive */
		if (xd+raw+best_comp > 1) {
			fprintf(stderr, "%s %s: Too many chunk options were selected.\nPlease select only one: -x, -r or -b\n", argv[0], argv[1]);
			poptPrintUsage(cdCon, stderr, 0);
			poptFreeContext(cdCon);
			return 1;
		}

		u32 mode = (raw ? RAW : (xd ? XDELTA : BEST_COMP));
		if (best_comp)
			gzip_level = MAX_GZIP_COMP;

		u64 start_addr = 0;
		if (resume && (!sscanf(resume, "%Lu", &start_addr))) {
			fprintf(stderr, "%s %s: Invalid resume position specified", argv[0], argv[1]);
			poptPrintUsage(cdCon, stderr, 0);
			poptFreeContext(cdCon);
			return 1;
		}

		u32 ratelimit = 0;
		if (ratelimit_str && ((ratelimit = strtobytes(ratelimit_str)) == INPUT_ERROR)) {
			fprintf(stderr, "Invalid rate limit input. Omit option, or use 0 for the default\n");
			poptPrintUsage(cdCon, stderr, 0);
			poptFreeContext(cdCon);
			return 1;
		}
		trace_off(fprintf(stderr, "xd=%d raw=%d best_comp=%d mode=%u gzip_level=%d\n", xd, raw, best_comp, mode, gzip_level););

		char const *sockname, *snaptag1str, *snaptag2str, *hoststr;

		sockname      = poptGetArg(cdCon);
		hoststr       = poptGetArg(cdCon);
		snaptag1str   = poptGetArg(cdCon);
		snaptag2str   = poptGetArg(cdCon);

		if ((sockname == NULL) || (hoststr == NULL) || (snaptag1str == NULL))
			cdUsage(cdCon, 1, argv[0], "Not enough arguments to send-delta\n");
		if (poptPeekArg(cdCon) != NULL)
			cdUsage(cdCon, 1, argv[0], "Too many arguments to send-delta\n");

		poptFreeContext(cdCon);

		char *hostname;
		unsigned port;
		hostname = strdup(hoststr);
		if (strchr(hostname, ':')) {
			unsigned int len = strlen(hostname);
			port = parse_port(hostname, &len);
			hostname[len] = '\0';
		} else {
			port = DEFAULT_REPLICATION_PORT;
		}
		int sock = create_socket(sockname);
		int ds_fd = open_socket(hostname, port);
		if (ds_fd < 0) {
			warn("%s %s: unable to connect to downstream server %s port %u: %s", argv[0], argv[1], hostname, port, strerror(errno));
			free(hostname);
			return 1;
		}
		free(hostname);

		u32 snaptag1, snaptag2;
		/* the fromsnap is optional. in case a single snap is specified, set src_snap to -1
		 * when calling ddsnap_replication_send to indicate full volume replication */
		if (parse_snaptag(snaptag1str, &snaptag1) < 0) {
			fprintf(stderr, "%s %s: invalid snapshot %s\n", argv[0], argv[1], snaptag1str);
			return 1;
		}
		if (snaptag2str == NULL) {
			snaptag2 = snaptag1;
			snaptag1 = -1;
		} else if (parse_snaptag(snaptag2str, &snaptag2) < 0) {
			fprintf(stderr, "%s %s: invalid snapshot %s\n", argv[0], argv[1], snaptag2str);
			return 1;
		}

		struct sigaction ign_sa;
		ign_sa.sa_handler = SIG_IGN;
		sigemptyset(&ign_sa.sa_mask);
		ign_sa.sa_flags = 0;
		if (sigaction(SIGPIPE, &ign_sa, NULL) == -1)
			warn("could not disable SIGPIPE: %s", strerror(errno));

		int ret;
		char *volume = strrchr(sockname, '/');
		if (!volume) {
			warn ("cannot get volume name from server sockname");
			ret = 1;
		} else {
			/* FIXME: should get device name from ddsnap server instead of assuming the naming style by zumastor */
			char *devstem = malloc(strlen(DEVMAP_PATH) + strlen(volume) + 1);
			if (!devstem) {
				warn ("cannot get volume name from server sockname");
				ret = 1;
			} else {
				sprintf(devstem, "%s%s", DEVMAP_PATH, volume);
				ret = ddsnap_replication_send(sock, snaptag1, snaptag2, devstem, mode, gzip_level, ds_fd, progress_file, start_addr, ratelimit);
				free(devstem);
			}
		}
		close(ds_fd);
		close(sock);

		return ret;
	}
	if (strcmp(command, "delta") == 0) {
		poptContext deltaCon;
		struct poptOption options[] = {
			{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &deltaOptions, 0, NULL, NULL },
			POPT_AUTOHELP
			POPT_TABLEEND
		};
		deltaCon = poptGetContext(NULL, argc-1, (const char **)&(argv[1]), options, 0);


		if (argc < 3) {
			poptPrintHelp(deltaCon, stdout, 0);
			poptFreeContext(deltaCon);
			exit(1);
		}

		char const *subcommand = argv[2];

		if (strcmp(subcommand, "--help") == 0 || strcmp(subcommand, "-?") == 0) {
			poptPrintHelp(deltaCon, stdout, 0);
			poptFreeContext(deltaCon);
			exit(0);
		}

		poptFreeContext(deltaCon);

		if (strcmp(subcommand, "--usage") == 0) {
			deltaUsage();
			exit(0);
		}

		if (strcmp(subcommand, "changelist") == 0) {
			if (argc != 7) {
				printf("usage: %s %s changelist <sockname> <changelist> <snapshot1> <snapshot2>\n", argv[0], argv[1]);
				return 1;
			}

			u32 snaptag1, snaptag2;

			if (parse_snaptag(argv[5], &snaptag1) < 0) {
				fprintf(stderr, "%s %s: invalid snapshot %s\n", argv[0], argv[1], argv[5]);
				return 1;
			}

			if (parse_snaptag(argv[6], &snaptag2) < 0) {
				fprintf(stderr, "%s %s: invalid snapshot %s\n", argv[0], argv[1], argv[6]);
				return 1;
			}

			int sock = create_socket(argv[3]);

			int ret = ddsnap_generate_changelist(sock, argv[4], snaptag1, snaptag2);
			close(sock);
			return ret;
		}
		if (strcmp(subcommand, "create") == 0) {
			char cdOpt;
			poptContext cdCon;

			struct poptOption options[] = {
				{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &cdOptions, 0, NULL, NULL },
				POPT_AUTOHELP
				POPT_TABLEEND
			};

			cdCon = poptGetContext(NULL, argc-2, (const char **)&(argv[2]), options, 0);
			poptSetOtherOptionHelp(cdCon, "<changelist> <deltafile> <devstem>");

			cdOpt = poptGetNextOpt(cdCon);

			if (argc < 3) {
				poptPrintUsage(cdCon, stderr, 0);
				exit(1);
			}

			if (cdOpt < -1) {
				/* an error occurred during option processing */
				fprintf(stderr, "%s: %s: %s\n",
					command,
					poptBadOption(cdCon, POPT_BADOPTION_NOALIAS),
					poptStrerror(cdOpt));
				poptFreeContext(cdCon);
				return 1;
			}

			/* Make sure the options are mutually exclusive */
			if (xd+raw+best_comp > 1) {
				fprintf(stderr, "%s %s: Too many chunk options were selected.\nPlease select only one: -x, -r or -b\n", argv[0], argv[1]);
				poptPrintUsage(cdCon, stderr, 0);
				poptFreeContext(cdCon);
				return 1;
			}

			u32 mode = (raw ? RAW : (xd? XDELTA : BEST_COMP));
			if (best_comp)
				gzip_level = MAX_GZIP_COMP;

			trace_off(fprintf(stderr, "xd=%d raw=%d best_comp=%d mode=%u gzip_level=%d\n", xd, raw, best_comp, mode, gzip_level););

			char const *changelist, *deltafile, *devstem;

			changelist = poptGetArg(cdCon);
			deltafile  = poptGetArg(cdCon);
			devstem    = poptGetArg(cdCon);

			if (changelist == NULL)
				cdUsage(cdCon, 1, "Specify a changelist", ".e.g., cl01\n");
			if (deltafile == NULL)
				cdUsage(cdCon, 1, "Specify a deltafile", ".e.g., df01\n");
			if (devstem == NULL)
				cdUsage(cdCon, 1, "Specify a devstem", ".e.g., /dev/mapper/snap\n");
			if (poptPeekArg(cdCon) != NULL)
				cdUsage(cdCon, 1, "Too many arguments inputted", "\n");

			int ret = ddsnap_generate_delta(mode, gzip_level, changelist, deltafile, devstem);

			poptFreeContext(cdCon);
			return ret;
		}
		if (strcmp(subcommand, "apply") == 0) {
			if (argc != 5) {
				printf("usage: %s %s apply <deltafile> <devstem>\n", argv[0], argv[1]);
				return 1;
			}
			return ddsnap_apply_delta(argv[3], argv[4]);
		}
		if (strcmp(subcommand, "listen") == 0) {
			char const *devstem;
			char const *hostspec;

			struct poptOption options[] = {
				{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &serverOptions, 0, NULL, NULL },
				POPT_AUTOHELP
				POPT_TABLEEND
			};

			poptContext dsCon = poptGetContext(NULL, argc-2, (const char **)&(argv[2]), options, 0);
			poptSetOtherOptionHelp(dsCon, "<devstem> <host>[:<port>]");

			char dsOpt = poptGetNextOpt(dsCon);

			if (dsOpt < -1) {
				fprintf(stderr, "%s %s: %s: %s\n", command, subcommand, poptBadOption(dsCon, POPT_BADOPTION_NOALIAS), poptStrerror(dsOpt));
				poptFreeContext(dsCon);
				return 1;
			}

			devstem = poptGetArg(dsCon);
			if (devstem == NULL) {
				fprintf(stderr, "%s %s: device stem must be specified\n", command, subcommand);
				poptPrintUsage(dsCon, stderr, 0);
				poptFreeContext(dsCon);
				return 1;
			}

			hostspec = poptGetArg(dsCon);

			if (poptPeekArg(dsCon) != NULL) {
				fprintf(stderr, "%s %s: only one host may be specified\n", command, subcommand);
				poptPrintUsage(dsCon, stderr, 0);
				poptFreeContext(dsCon);
				return 1;
			}

			poptFreeContext(dsCon);

			char *hostname;
			unsigned port;

			if (!hostspec) {
				hostname = strdup("0.0.0.0");
				port = DEFAULT_REPLICATION_PORT;
			} else {
				hostname = strdup(hostspec);

				if (strchr(hostname, ':')) {
					unsigned int len = strlen(hostname);
					port = parse_port(hostname, &len);
					hostname[len] = '\0';
				} else {
					port = DEFAULT_REPLICATION_PORT;
				}
			}

			/* make sure origin device exists (catch typos early) */
			int origin = open(devstem, O_RDONLY);
			if (origin < 0) {
				fprintf(stderr, "%s %s: unable to open origin device \"%s\" for reading: %s\n", command, subcommand, devstem, strerror(errno));
				free(hostname);
				return 1;
			}

			close(origin);

			int sock = bind_socket(hostname, port);
			if (sock < 0) {
				fprintf(stderr, "%s %s: unable to bind to %s port %u, err: %s\n", command, subcommand, hostname, port, strerror(-sock));
				free(hostname);
				return 1;
			}
			free(hostname);

			int getsigfd;
			if (!nobg) {
				pid_t pid;

				if (!logfile)
					logfile = "/var/log/ddsnap.delta.log";

				pid = daemonize(logfile, pidfile, &getsigfd);
				if (pid == -1)
					error("Error: could not daemonize\n");
				if (pid != 0) {
					trace_off(printf("pid = %lu\n", (unsigned long)pid););
					return 0;
				}
			}


			return ddsnap_delta_server(sock, devstem, progress_file,
				logfile, getsigfd);
		}

		fprintf(stderr, "%s %s: unrecognized delta subcommand: %s.\n", argv[0], command, subcommand);
		return 1;
	}
	if (strcmp(command, "resize") == 0) {
		struct poptOption options[] = {
			{ NULL, '\0', POPT_ARG_INCLUDE_TABLE, &resizeOptions, 0, NULL, NULL },
			POPT_AUTOHELP
			POPT_TABLEEND
		};

		poptContext cdCon = poptGetContext(NULL, argc-1, (const char **)&(argv[1]), options, 0);
		poptSetOtherOptionHelp(cdCon, "<server_socket>");

		char cdOpt = poptGetNextOpt(cdCon);
		if (cdOpt < -1) {
			fprintf(stderr, "%s: %s: %s\n", argv[0], poptBadOption(cdCon, POPT_BADOPTION_NOALIAS), poptStrerror(cdOpt));
			poptFreeContext(cdCon);
			return 1;
		}

		u64 orgsize=0, snapsize=0, metasize=0;
		if (orgsize_str != NULL)
			orgsize = strtobytes64(orgsize_str);
		if (snapsize_str != NULL)
			snapsize = strtobytes64(snapsize_str);
		if (metasize_str != NULL)
			metasize = strtobytes64(metasize_str);
		if (orgsize == INPUT_ERROR_64 || snapsize == INPUT_ERROR_64 || metasize == INPUT_ERROR_64) {
			fprintf(stderr, "ddsnap: Invalid size input. Omit option, or use 0 for keep the old size\n");
			return 1;
		}
		
		const char *sockname = poptGetArg(cdCon);
		if (sockname == NULL)
			cdUsage(cdCon, 1, argv[0], "Must specify socket name to resize\n");
		if (poptPeekArg(cdCon) != NULL)
			cdUsage(cdCon, 1, argv[0], "Too many arguments to resize\n");

		poptFreeContext(cdCon);

		int sock = create_socket(sockname);

		int ret = ddsnap_resize_devices(sock, orgsize, snapsize, metasize);

		return ret;
	}
	/* syntax for snapshot revert: ddsnap revert <socket> <snap> */
	if (strcmp(command, "revert") == 0) {
		u32 snaptag;
		int err = 0, orgdev = -1, snapdev = -1, serv_fd;
		char *devstem = NULL, *devname = NULL, *buffer = NULL, *volume;
		struct change_list *cl = NULL;

		if (argc != 4) {
			printf("Usage: %s revert <sockname> <snapshot>\n", argv[0]);
			return -1;
		}
		if (parse_snaptag(argv[3], &snaptag) < 0) {
			fprintf(stderr, "%s: invalid snapshot %s\n", argv[0], argv[3]);
			return -1;
		}
		serv_fd = create_socket(argv[2]);

		err = -EINVAL;
		if (!(volume = strrchr(argv[2], '/'))) {
			warn("cannot get volume name from server sockname");
			goto out;
		}
		if ((cl = stream_changelist(serv_fd, (u32)~0UL, snaptag)) == NULL) {
			warn("could not receive change list between origin and snapshot %Lu", (llu_t) snaptag);
			goto out;
		}

		err = -ENOMEM;
		size_t chunksize = 1 << cl->chunksize_bits;
		if (!(buffer = malloc(chunksize)))
			goto out;
		if (!(devstem = malloc(strlen(DEVMAP_PATH) + strlen(volume) + 1)))
			goto out;
		sprintf(devstem, "%s%s", DEVMAP_PATH, volume);
		if (!(devname = malloc_snapshot_name(devstem, snaptag)))
			goto out;

		trace_off(printf("volume %s, devstem %s, devname %s\n", volume, devstem, devname););
		if ((orgdev = open(devstem, O_WRONLY)) < 0) {
			warn("unable to open origin device %s: %s", devstem, strerror(errno));
			err = orgdev;
			goto out;
		}
		if ((snapdev = open(devname, O_RDONLY)) < 0) {
			warn("unable to open target snapshot: %s", strerror(errno));
			err = snapdev;
			goto out;
		}

		err = -EIO;
		trace_off(printf("cl->count %Lu, chunksize %u\n", cl->count, chunksize););
		for (int i = 0; i < cl->count; i++) {
			off_t chunkpos = cl->chunks[i] << cl->chunksize_bits;
			if (diskread(snapdev, buffer, chunksize, chunkpos) == -1) {
				warn("error when reading snapshot at %Lu", (unsigned long long)chunkpos);
				goto out;
			}
			if (diskwrite(orgdev, buffer, chunksize, chunkpos) == -1) {
				warn("error when writing origin at %Lu", (unsigned long long)chunkpos);<