far-rcv/far-rcv.c

/*
 * Copyright (C) 2012 Alexander Block.
 * Copyright (C) 2012, 2013 STRATO AG.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 * DEALINGS IN THE SOFTWARE.
 */

/*
 * This is the main part of the far-rcv library. All exported functions
 * are located in this file.
 */

#define _BSD_SOURCE
#define _GNU_SOURCE
#define _ATFILE_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <assert.h>
#include <stdint.h>
#include <inttypes.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <netinet/in.h>
#if !defined (HAVE_UTIMENSAT)
#include <utime.h>
#endif /* !defined (HAVE_UTIMENSAT) */

#include "far-rcv.h"
#include "far-crc32c.h"
#include "far-endian.h"
#include "far-xattr.h"


#if !defined (O_NOATIME)
#define O_NOATIME 0
#endif


#define __TLV_GOTO_FAIL(expr) \
	if ((ret = expr) < 0) \
		goto tlv_get_failed;

#define __TLV_DO_WHILE_GOTO_FAIL(expr) \
	do { \
		__TLV_GOTO_FAIL(expr) \
	} while (0)


#define TLV_GET(s, attr, data, len) \
	__TLV_DO_WHILE_GOTO_FAIL(tlv_get(s, attr, data, len))

#define TLV_CHECK_LEN(expected, got) \
	do { \
		if (expected != got) { \
			fprintf(stderr, \
				"ERROR: invalid size for attribute. expected = %d, got = %d\n", \
				(int)expected, (int)got); \
			ret = -EINVAL; \
			goto tlv_get_failed; \
		} \
	} while (0)

#define TLV_GET_INT(s, attr, bits, v) \
	do { \
		uint##bits##_t __tmp; \
		void *__ptr; \
		int __len; \
		TLV_GET(s, attr, &__ptr, &__len); \
		TLV_CHECK_LEN(sizeof(__tmp), __len); \
		memcpy(&__tmp, __ptr, sizeof(__tmp)); \
		*v = far_rcv_le##bits##toh(__tmp); \
	} while (0)

#define far_rcv_le8toh(v) (v)
#define TLV_GET_U8(s, attr, v) TLV_GET_INT(s, attr, 8, v)
#define TLV_GET_U16(s, attr, v) TLV_GET_INT(s, attr, 16, v)
#define TLV_GET_U32(s, attr, v) TLV_GET_INT(s, attr, 32, v)
#define TLV_GET_U64(s, attr, v) TLV_GET_INT(s, attr, 64, v)

#define TLV_GET_STRING(s, attr, str) \
	__TLV_DO_WHILE_GOTO_FAIL(tlv_get_string(s, attr, str))

#define TLV_GET_TIMESPEC(s, attr, ts) \
	__TLV_DO_WHILE_GOTO_FAIL(tlv_get_timespec(s, attr, ts))

#define TLV_GET_UUID(s, attr, uuid) \
	__TLV_DO_WHILE_GOTO_FAIL(tlv_get_uuid(s, attr, uuid))


#define FAR_SEND_STREAM_MAGIC "far-stream\0\0"
#define FAR_SEND_STREAM_MAGIC_ALT "btrfs-stream"
#define FAR_SEND_STREAM_VERSION 1

#define FAR_SEND_BUF_SIZE (1024 * 64)
#define FAR_SEND_READ_SIZE (1024 * 48)

enum far_tlv_type {
	FAR_TLV_U8,
	FAR_TLV_U16,
	FAR_TLV_U32,
	FAR_TLV_U64,
	FAR_TLV_BINARY,
	FAR_TLV_STRING,
	FAR_TLV_UUID,
	FAR_TLV_TIMESPEC,
};

struct far_timespec {
	uint64_t sec;
	uint32_t nsec;
} __attribute__ ((__packed__));

struct far_stream_header {
	char magic[sizeof(FAR_SEND_STREAM_MAGIC)];
	uint32_t version;
} __attribute__ ((__packed__));

struct far_cmd_header {
	/* len excluding the header */
	uint32_t len;
	uint16_t cmd;
	/* crc including the header with zero crc field */
	uint32_t crc;
} __attribute__ ((__packed__));

struct far_tlv_header {
	uint16_t tlv_type;
	/* len excluding the header */
	uint16_t tlv_len;
} __attribute__ ((__packed__));

/* commands */
enum far_send_cmd {
	FAR_SEND_C_UNSPEC,

	FAR_SEND_C_SUBVOL,
	FAR_SEND_C_SNAPSHOT,

	FAR_SEND_C_MKFILE,
	FAR_SEND_C_MKDIR,
	FAR_SEND_C_MKNOD,
	FAR_SEND_C_MKFIFO,
	FAR_SEND_C_MKSOCK,
	FAR_SEND_C_SYMLINK,

	FAR_SEND_C_RENAME,
	FAR_SEND_C_LINK,
	FAR_SEND_C_UNLINK,
	FAR_SEND_C_RMDIR,

	FAR_SEND_C_SET_XATTR,
	FAR_SEND_C_REMOVE_XATTR,

	FAR_SEND_C_WRITE,
	FAR_SEND_C_CLONE,

	FAR_SEND_C_TRUNCATE,
	FAR_SEND_C_CHMOD,
	FAR_SEND_C_CHOWN,
	FAR_SEND_C_UTIMES,

	FAR_SEND_C_END,
	__FAR_SEND_C_MAX,
};
#define FAR_SEND_C_MAX (__FAR_SEND_C_MAX - 1)

/* attributes in send stream */
enum {
	FAR_SEND_A_UNSPEC,

	FAR_SEND_A_UUID,
	FAR_SEND_A_CTRANSID,

	FAR_SEND_A_INO,
	FAR_SEND_A_SIZE,
	FAR_SEND_A_MODE,
	FAR_SEND_A_UID,
	FAR_SEND_A_GID,
	FAR_SEND_A_RDEV,
	FAR_SEND_A_CTIME,
	FAR_SEND_A_MTIME,
	FAR_SEND_A_ATIME,
	FAR_SEND_A_OTIME,

	FAR_SEND_A_XATTR_NAME,
	FAR_SEND_A_XATTR_DATA,

	FAR_SEND_A_PATH,
	FAR_SEND_A_PATH_TO,
	FAR_SEND_A_PATH_LINK,

	FAR_SEND_A_FILE_OFFSET,
	FAR_SEND_A_DATA,

	FAR_SEND_A_CLONE_UUID,
	FAR_SEND_A_CLONE_CTRANSID,
	FAR_SEND_A_CLONE_PATH,
	FAR_SEND_A_CLONE_OFFSET,
	FAR_SEND_A_CLONE_LEN,

	__FAR_SEND_A_MAX,
};
#define FAR_SEND_A_MAX (__FAR_SEND_A_MAX - 1)

struct far_send_stream {
	int fd;
	char read_buf[FAR_SEND_BUF_SIZE];

	int cmd;
	struct far_tlv_header *cmd_attrs[FAR_SEND_A_MAX + 1];
	uint32_t version;
};


static int read_and_process_send_stream(struct far_rcv_ctx *frctx, int fd);
static int read_buf(struct far_send_stream *s, void *buf, int len);
static int read_cmd(struct far_send_stream *s);
static int tlv_get(struct far_send_stream *s, int attr, void **data, int *len);
static int tlv_get_string(struct far_send_stream *s, int attr, char **str);
static int tlv_get_timespec(struct far_send_stream *s, int attr,
			    struct timespec *ts);
static int tlv_get_uuid(struct far_send_stream *s, int attr,
			unsigned char *uuid);
static int read_and_process_cmd(struct far_rcv_ctx *frctx,
				struct far_send_stream *s);
static int process_mkfile(struct far_rcv_ctx *frctx, const char *path);
static int process_mkdir(struct far_rcv_ctx *frctx, const char *path);
static int process_mknod(struct far_rcv_ctx *frctx, const char *path,
			 uint64_t mode, uint64_t dev);
static int process_mkfifo(struct far_rcv_ctx *frctx, const char *path);
static int process_mksock(struct far_rcv_ctx *frctx, const char *path);
static int process_symlink(struct far_rcv_ctx *frctx, const char *path,
			   const char *lnk);
static int process_rename(struct far_rcv_ctx *frctx, const char *from,
			  const char *to);
static int process_link(struct far_rcv_ctx *frctx, const char *path,
			const char *lnk);
static int process_unlink(struct far_rcv_ctx *frctx, const char *path);
static int process_rmdir(struct far_rcv_ctx *frctx, const char *path);
static int open_inode_for_write(struct far_rcv_ctx *frctx, const char *path);
static int close_inode_for_write(struct far_rcv_ctx *frctx);
static int process_write(struct far_rcv_ctx *frctx, const char *path,
			 const void *data, uint64_t offset, uint64_t len);
static int process_clone(struct far_rcv_ctx *frctx, const char *path,
			 uint64_t offset, uint64_t len,
			 const unsigned char *clone_uuid,
			 uint64_t clone_ctransid, const char *clone_path,
			 uint64_t clone_offset);
static int process_set_xattr(struct far_rcv_ctx *frctx, const char *path,
			     const char *name, const void *data, int len);
static int process_remove_xattr(struct far_rcv_ctx *frctx, const char *path,
				const char *name);
static int process_truncate(struct far_rcv_ctx *frctx, const char *path,
			    uint64_t size);
static int process_chmod(struct far_rcv_ctx *frctx, const char *path,
			 uint64_t mode);
static int process_chown(struct far_rcv_ctx *frctx, const char *path,
			 uint64_t uid, uint64_t gid);
static int process_utimes(struct far_rcv_ctx *frctx, const char *path,
			  struct timespec *at, struct timespec *mt,
			  struct timespec *ct);


int far_rcv_init(struct far_rcv_ctx *frctx)
{
	assert(frctx);

	memset(frctx, 0, sizeof(*frctx));
	frctx->verbose = 0;
	frctx->support_xattrs = 1;
	frctx->free_current_base_path = 0;
	frctx->current_base_path = NULL;
	frctx->write_fd = -1;
	frctx->write_path = NULL;

	frctx->ops.finish_subvol = NULL;
	frctx->ops.subvol = NULL;
	frctx->ops.snapshot = NULL;
	frctx->ops.mkfile = process_mkfile;
	frctx->ops.mkdir = process_mkdir;
	frctx->ops.mknod = process_mknod;
	frctx->ops.mkfifo = process_mkfifo;
	frctx->ops.mksock = process_mksock;
	frctx->ops.symlink = process_symlink;
	frctx->ops.rename = process_rename;
	frctx->ops.link = process_link;
	frctx->ops.unlink = process_unlink;
	frctx->ops.rmdir = process_rmdir;
	frctx->ops.open_inode_for_write = open_inode_for_write;
	frctx->ops.close_inode_for_write = close_inode_for_write;
	frctx->ops.write = process_write;
	frctx->ops.clone = process_clone;
	frctx->ops.set_xattr = process_set_xattr;
	frctx->ops.remove_xattr = process_remove_xattr;
	frctx->ops.truncate = process_truncate;
	frctx->ops.chmod = process_chmod;
	frctx->ops.chown = process_chown;
	frctx->ops.utimes = process_utimes;

	return 0;
}

void far_rcv_deinit(struct far_rcv_ctx *frctx)
{
	if (frctx->free_current_base_path) {
		free((void *)frctx->current_base_path);
		frctx->free_current_base_path = 0;
	}
	frctx->current_base_path = NULL;
	frctx->root_path = NULL;
	free((void *)frctx->write_path);
	frctx->write_path = NULL;
	if (frctx->write_fd != -1) {
		close(frctx->write_fd);
		frctx->write_fd = -1;
	}
}

int far_rcv_mainloop(struct far_rcv_ctx *frctx, int stream_fd,
		     const char *root_path)
{
	int end = 0;
	int ret;
	frctx->root_path = root_path;
	frctx->current_base_path = root_path;
	frctx->free_current_base_path = 0;

	while (!end) {
		ret = read_and_process_send_stream(frctx, stream_fd);
		if (ret < 0)
			goto out;
		if (ret)
			end = 1;

		ret = frctx->ops.close_inode_for_write(frctx);
		if (ret < 0)
			goto out;
		ret = frctx->ops.finish_subvol(frctx);
		if (ret < 0)
			goto out;
	}
	ret = 0;

out:
	far_rcv_deinit(frctx);
	return ret;
}

static int read_and_process_send_stream(struct far_rcv_ctx *frctx, int fd)
{
	int ret;
	struct far_send_stream s;
	struct far_stream_header hdr;

	s.fd = fd;

	ret = read_buf(&s, &hdr, sizeof(hdr));
	if (ret < 0)
		goto out;
	if (ret) {
		ret = 1;
		goto out;
	}

	if (strcmp(hdr.magic, FAR_SEND_STREAM_MAGIC) &&
	    strcmp(hdr.magic, FAR_SEND_STREAM_MAGIC_ALT)) {
		ret = -EINVAL;
		fprintf(stderr, "ERROR: Unexpected header\n");
		goto out;
	}

	s.version = far_rcv_le32toh(hdr.version);
	if (s.version > FAR_SEND_STREAM_VERSION) {
		ret = -EINVAL;
		fprintf(stderr,
			"ERROR: Stream version %d not supported. Please upgrade.\n",
			s.version);
		goto out;
	}

	while (1) {
		ret = read_and_process_cmd(frctx, &s);
		if (ret < 0)
			goto out;
		if (ret) {
			/* received end marker */
			goto out;
		}
	}

out:
	return ret;
}

static int read_buf(struct far_send_stream *s, void *buf, int len)
{
	int ret;
	int pos = 0;

	while (pos < len) {
		ret = read(s->fd, (char*)buf + pos, len - pos);
		if (ret < 0) {
			ret = -errno;
			fprintf(stderr, "ERROR: read from stream failed. %s\n",
				strerror(-ret));
			goto out;
		}
		if (ret == 0) {
			ret = 1;
			goto out;
		}
		pos += ret;
	}

	ret = 0;

out:
	return ret;
}

/*
 * Reads a single command and decodes the TLV's into s->cmd_attrs
 */
static int read_cmd(struct far_send_stream *s)
{
	int ret;
	int cmd;
	int cmd_len;
	int tlv_type;
	int tlv_len;
	char *data;
	int pos;
	struct far_tlv_header tlv_hdr;
	struct far_cmd_header cmd_hdr;
	uint32_t crc;
	uint32_t crc2;

	memset(s->cmd_attrs, 0, sizeof(s->cmd_attrs));

	ret = read_buf(s, &cmd_hdr, sizeof(cmd_hdr));
	if (ret < 0)
		goto out;
	if (ret) {
		ret = -EINVAL;
		fprintf(stderr, "ERROR: unexpected EOF in stream.\n");
		goto out;
	}

	cmd = far_rcv_le16toh(cmd_hdr.cmd);
	cmd_len = far_rcv_le32toh(cmd_hdr.len);
	crc = far_rcv_le32toh(cmd_hdr.crc);
	cmd_hdr.crc = 0;
	crc2 = far_crc32c_calc(0, (unsigned char*)&cmd_hdr, sizeof(cmd_hdr));

	ret = read_buf(s, s->read_buf, cmd_len);
	if (ret < 0)
		goto out;
	if (ret) {
		ret = -EINVAL;
		fprintf(stderr, "ERROR: unexpected EOF in stream.\n");
		goto out;
	}

	crc2 = far_crc32c_calc(crc2, (unsigned char*)s->read_buf, cmd_len);

	if (crc != crc2) {
		ret = -EINVAL;
		fprintf(stderr, "ERROR: crc32 mismatch in command.\n");
		goto out;
	}

	pos = 0;
	data = s->read_buf;
	while (pos < cmd_len) {
		memcpy(&tlv_hdr, data, sizeof(tlv_hdr));
		tlv_type = far_rcv_le16toh(tlv_hdr.tlv_type);
		tlv_len = far_rcv_le16toh(tlv_hdr.tlv_len);

		if (tlv_type <= 0 || tlv_type > FAR_SEND_A_MAX ||
		    tlv_len < 0 || tlv_len > FAR_SEND_BUF_SIZE) {
			fprintf(stderr,
				"ERROR: invalid tlv in cmd. tlv_type = %d, tlv_len = %d\n",
				tlv_type, tlv_len);
			ret = -EINVAL;
			goto out;
		}

		s->cmd_attrs[tlv_type] = (struct far_tlv_header *)data;
		data += sizeof(tlv_hdr) + tlv_len;
		pos += sizeof(tlv_hdr) + tlv_len;
	}

	s->cmd = cmd;
	ret = 0;

out:
	return ret;
}

static int tlv_get(struct far_send_stream *s, int attr, void **data, int *len)
{
	int ret;
	struct far_tlv_header tlv_hdr;
	struct far_tlv_header *h;

	if (attr <= 0 || attr > FAR_SEND_A_MAX) {
		fprintf(stderr,
			"ERROR: invalid attribute requested. attr = %d\n",
			attr);
		ret = -EINVAL;
		goto out;
	}

	h = s->cmd_attrs[attr];
	if (!h) {
		fprintf(stderr,
			"ERROR: attribute %d requested but not present.\n",
			attr);
		ret = -ENOENT;
		goto out;
	}

	memcpy(&tlv_hdr, h, sizeof(tlv_hdr));
	*len = far_rcv_le16toh(tlv_hdr.tlv_len);
	*data = h + 1;

	ret = 0;

out:
	return ret;
}

static int tlv_get_string(struct far_send_stream *s, int attr, char **str)
{
	int ret;
	void *data;
	int len;

	TLV_GET(s, attr, &data, &len);

	*str = malloc(len + 1);
	if (!*str)
		return -ENOMEM;

	memcpy(*str, data, len);
	(*str)[len] = 0;
	ret = 0;

tlv_get_failed:
	return ret;
}

static int tlv_get_timespec(struct far_send_stream *s, int attr,
			    struct timespec *ts)
{
	int ret;
	int len;
	struct far_timespec bts;
	void *ptr;

	TLV_GET(s, attr, &ptr, &len);
	TLV_CHECK_LEN(sizeof(bts), len);

	memcpy(&bts, ptr, sizeof(bts));
	ts->tv_sec = far_rcv_le64toh(bts.sec);
	ts->tv_nsec = far_rcv_le32toh(bts.nsec);
	ret = 0;

tlv_get_failed:
	return ret;
}

static int tlv_get_uuid(struct far_send_stream *s, int attr,
			unsigned char *uuid)
{
	int ret;
	int len;
	void *data;

	TLV_GET(s, attr, &data, &len);
	TLV_CHECK_LEN(FAR_UUID_SIZE, len);
	memcpy(uuid, data, FAR_UUID_SIZE);

	ret = 0;

tlv_get_failed:
	return ret;
}

static int read_and_process_cmd(struct far_rcv_ctx *frctx,
				struct far_send_stream *s)
{
	int ret;
	char *path = NULL;
	char *path_to = NULL;
	char *clone_path = NULL;
	char *xattr_name = NULL;
	void *xattr_data = NULL;
	void *data = NULL;
	struct timespec at;
	struct timespec ct;
	struct timespec mt;
	unsigned char uuid[FAR_UUID_SIZE];
	unsigned char clone_uuid[FAR_UUID_SIZE];
	uint64_t tmp;
	uint64_t tmp2;
	uint64_t ctransid;
	uint64_t clone_ctransid;
	uint64_t mode;
	uint64_t dev;
	uint64_t clone_offset;
	uint64_t offset;
	int len;
	int xattr_len;
	char *tmp_path;

	ret = read_cmd(s);
	if (ret)
		goto out;

	switch (s->cmd) {
	case FAR_SEND_C_SUBVOL:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_UUID(s, FAR_SEND_A_UUID, uuid);
		TLV_GET_U64(s, FAR_SEND_A_CTRANSID, &ctransid);
		tmp_path = strrchr(path, '/');
		if (tmp_path)
			path = strdup(tmp_path + 1);
		ret = frctx->ops.subvol(frctx, path, uuid, ctransid);
		break;
	case FAR_SEND_C_SNAPSHOT:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_UUID(s, FAR_SEND_A_UUID, uuid);
		TLV_GET_U64(s, FAR_SEND_A_CTRANSID, &ctransid);
		TLV_GET_UUID(s, FAR_SEND_A_CLONE_UUID, clone_uuid);
		TLV_GET_U64(s, FAR_SEND_A_CLONE_CTRANSID, &clone_ctransid);
		tmp_path = strrchr(path, '/');
		if (tmp_path)
			path = strdup(tmp_path + 1);
		ret = frctx->ops.snapshot(frctx, path, uuid, ctransid,
					  clone_uuid, clone_ctransid);
		break;
	case FAR_SEND_C_MKFILE:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		ret = frctx->ops.mkfile(frctx, path);
		break;
	case FAR_SEND_C_MKDIR:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		ret = frctx->ops.mkdir(frctx, path);
		break;
	case FAR_SEND_C_MKNOD:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_U64(s, FAR_SEND_A_MODE, &mode);
		TLV_GET_U64(s, FAR_SEND_A_RDEV, &dev);
		ret = frctx->ops.mknod(frctx, path, mode, dev);
		break;
	case FAR_SEND_C_MKFIFO:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		ret = frctx->ops.mkfifo(frctx, path);
		break;
	case FAR_SEND_C_MKSOCK:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		ret = frctx->ops.mksock(frctx, path);
		break;
	case FAR_SEND_C_SYMLINK:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_STRING(s, FAR_SEND_A_PATH_LINK, &path_to);
		ret = frctx->ops.symlink(frctx, path, path_to);
		break;
	case FAR_SEND_C_RENAME:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_STRING(s, FAR_SEND_A_PATH_TO, &path_to);
		ret = frctx->ops.rename(frctx, path, path_to);
		break;
	case FAR_SEND_C_LINK:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_STRING(s, FAR_SEND_A_PATH_LINK, &path_to);
		ret = frctx->ops.link(frctx, path, path_to);
		break;
	case FAR_SEND_C_UNLINK:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		ret = frctx->ops.unlink(frctx, path);
		break;
	case FAR_SEND_C_RMDIR:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		ret = frctx->ops.rmdir(frctx, path);
		break;
	case FAR_SEND_C_WRITE:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_U64(s, FAR_SEND_A_FILE_OFFSET, &offset);
		TLV_GET(s, FAR_SEND_A_DATA, &data, &len);
		ret = frctx->ops.write(frctx, path, data, offset, len);
		break;
	case FAR_SEND_C_CLONE:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_U64(s, FAR_SEND_A_FILE_OFFSET, &offset);
		TLV_GET_U64(s, FAR_SEND_A_CLONE_LEN, &len);
		TLV_GET_UUID(s, FAR_SEND_A_CLONE_UUID, clone_uuid);
		TLV_GET_U64(s, FAR_SEND_A_CLONE_CTRANSID, &clone_ctransid);
		TLV_GET_STRING(s, FAR_SEND_A_CLONE_PATH, &clone_path);
		TLV_GET_U64(s, FAR_SEND_A_CLONE_OFFSET, &clone_offset);
		ret = frctx->ops.clone(frctx, path, offset, len, clone_uuid,
				       clone_ctransid, clone_path,
				       clone_offset);
		break;
	case FAR_SEND_C_SET_XATTR:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_STRING(s, FAR_SEND_A_XATTR_NAME, &xattr_name);
		TLV_GET(s, FAR_SEND_A_XATTR_DATA, &xattr_data, &xattr_len);
		if (!frctx->support_xattrs)
			break;
		ret = frctx->ops.set_xattr(frctx, path, xattr_name, xattr_data,
					   xattr_len);
		break;
	case FAR_SEND_C_REMOVE_XATTR:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_STRING(s, FAR_SEND_A_XATTR_NAME, &xattr_name);
		if (!frctx->support_xattrs)
			break;
		ret = frctx->ops.remove_xattr(frctx, path, xattr_name);
		break;
	case FAR_SEND_C_TRUNCATE:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_U64(s, FAR_SEND_A_SIZE, &tmp);
		ret = frctx->ops.truncate(frctx, path, tmp);
		break;
	case FAR_SEND_C_CHMOD:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_U64(s, FAR_SEND_A_MODE, &tmp);
		ret = frctx->ops.chmod(frctx, path, tmp);
		break;
	case FAR_SEND_C_CHOWN:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_U64(s, FAR_SEND_A_UID, &tmp);
		TLV_GET_U64(s, FAR_SEND_A_GID, &tmp2);
		ret = frctx->ops.chown(frctx, path, tmp, tmp2);
		break;
	case FAR_SEND_C_UTIMES:
		TLV_GET_STRING(s, FAR_SEND_A_PATH, &path);
		TLV_GET_TIMESPEC(s, FAR_SEND_A_ATIME, &at);
		TLV_GET_TIMESPEC(s, FAR_SEND_A_MTIME, &mt);
		TLV_GET_TIMESPEC(s, FAR_SEND_A_CTIME, &ct);
		ret = frctx->ops.utimes(frctx, path, &at, &mt, &ct);
		break;
	case FAR_SEND_C_END:
		if (frctx->verbose >= 2)
			fprintf(stderr, "receive end marker\n");
		ret = 1;
		break;
	}

tlv_get_failed:
out:
	free(path);
	free(path_to);
	free(clone_path);
	free(xattr_name);
	return ret;
}

static int process_mkfile(struct far_rcv_ctx *frctx, const char *path)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "mkfile %s\n", path);

	ret = creat(full_path, 0600);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: mkfile %s failed. %s\n", path,
			strerror(-ret));
		goto out;
	}
	close(ret);
	ret = 0;

out:
	free(full_path);
	return ret;
}

static int process_mkdir(struct far_rcv_ctx *frctx, const char *path)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "mkdir %s\n", path);

	ret = mkdir(full_path, 0700);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: mkdir %s failed. %s\n", path,
			strerror(-ret));
	}

	free(full_path);
	return ret;
}

static int process_mknod(struct far_rcv_ctx *frctx, const char *path,
			 uint64_t mode, uint64_t dev)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "mknod %s mode=%" PRIu64 ", dev=%" PRIu64 "\n",
			path, mode, dev);

	ret = mknod(full_path, mode & S_IFMT, dev);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: mknod %s failed. %s\n", path,
			strerror(-ret));
	}

	free(full_path);
	return ret;
}

static int process_mkfifo(struct far_rcv_ctx *frctx, const char *path)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "mkfifo %s\n", path);

	ret = mkfifo(full_path, 0600);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: mkfifo %s failed. %s\n", path,
			strerror(-ret));
	}

	free(full_path);
	return ret;
}

static int process_mksock(struct far_rcv_ctx *frctx, const char *path)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "mksock %s\n", path);

	ret = mknod(full_path, 0600 | S_IFSOCK, 0);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: mknod %s failed. %s\n", path,
			strerror(-ret));
	}

	free(full_path);
	return ret;
}

static int process_symlink(struct far_rcv_ctx *frctx, const char *path,
			   const char *lnk)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "symlink %s -> %s\n", path, lnk);

	ret = symlink(lnk, full_path);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: symlink %s -> %s failed. %s\n", path,
			lnk, strerror(-ret));
	}

	free(full_path);
	return ret;
}

static int process_rename(struct far_rcv_ctx *frctx, const char *from,
			  const char *to)
{
	int ret;
	char *full_from = far_rcv_path_cat(frctx->current_base_path, from);
	char *full_to = far_rcv_path_cat(frctx->current_base_path, to);

	if (frctx->verbose >= 2)
		fprintf(stderr, "rename %s -> %s\n", from, to);

	ret = rename(full_from, full_to);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: rename %s -> %s failed. %s\n", from,
			to, strerror(-ret));
	}

	free(full_from);
	free(full_to);
	return ret;
}

static int process_link(struct far_rcv_ctx *frctx, const char *path,
			const char *lnk)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);
	char *full_link_path = far_rcv_path_cat(frctx->current_base_path, lnk);

	if (frctx->verbose >= 2)
		fprintf(stderr, "link %s -> %s\n", path, lnk);

	ret = link(full_link_path, full_path);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: link %s -> %s failed. %s\n", path,
			lnk, strerror(-ret));
	}

	free(full_path);
	free(full_link_path);
	return ret;
}


static int process_unlink(struct far_rcv_ctx *frctx, const char *path)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "unlink %s\n", path);

	ret = unlink(full_path);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: unlink %s failed. %s\n", path,
			strerror(-ret));
	}

	free(full_path);
	return ret;
}

static int process_rmdir(struct far_rcv_ctx *frctx, const char *path)
{
	int ret;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "rmdir %s\n", path);

	ret = rmdir(full_path);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: rmdir %s failed. %s\n", path,
			strerror(-ret));
	}

	free(full_path);
	return ret;
}

static int open_inode_for_write(struct far_rcv_ctx *frctx, const char *path)
{
	int ret = 0;

	if (frctx->write_fd != -1) {
		if (strcmp(frctx->write_path, path) == 0)
			goto out;
		close(frctx->write_fd);
		frctx->write_fd = -1;
	}

	frctx->write_fd = open(path, O_RDWR);
	if (frctx->write_fd < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: open %s failed. %s\n", path,
			strerror(-ret));
		goto out;
	}
	free((void *)frctx->write_path);
	frctx->write_path = strdup(path);

out:
	return ret;
}

static int close_inode_for_write(struct far_rcv_ctx *frctx)
{
	free((void *)frctx->write_path);
	frctx->write_path = NULL;
	if (frctx->write_fd != -1) {
		close(frctx->write_fd);
		frctx->write_fd = -1;
	}

	return 0;
}

static int process_write(struct far_rcv_ctx *frctx, const char *path,
			 const void *data, uint64_t offset, uint64_t len)
{
	int ret = 0;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);
	uint64_t pos = 0;
	int w;

	ret = frctx->ops.open_inode_for_write(frctx, full_path);
	if (ret < 0)
		goto out;

	while (pos < len) {
		w = pwrite(frctx->write_fd, (char*)data + pos, len - pos,
			   offset + pos);
		if (w < 0) {
			ret = -errno;
			fprintf(stderr, "ERROR: writing to %s failed. %s\n",
				path, strerror(-ret));
			goto out;
		}
		pos += w;
	}

out:
	free(full_path);
	return ret;
}

static int process_clone(struct far_rcv_ctx *frctx, const char *path,
			 uint64_t offset, uint64_t len,
			 const unsigned char *clone_uuid,
			 uint64_t clone_ctransid, const char *clone_path,
			 uint64_t clone_offset)
{
	int ret;
	char *full_dst_path = far_rcv_path_cat(frctx->current_base_path, path);
	char *full_clone_path = NULL;
	int clone_fd = -1;

	ret = frctx->ops.open_inode_for_write(frctx, full_dst_path);
	if (ret < 0)
		goto out;

	full_clone_path = far_rcv_path_cat(frctx->current_base_path,
					   clone_path);

	clone_fd = open(full_clone_path, O_RDONLY | O_NOATIME);
	if (clone_fd < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: failed to open %s. %s\n",
			full_clone_path, strerror(-ret));
		goto out;
	}

	if (lseek(frctx->write_fd, offset, SEEK_SET) == (off_t)-1) {
		ret = -errno;
		fprintf(stderr, "ERROR: failed to lseek %s. %s\n",
			full_dst_path, strerror(-ret));
		goto out;
	}
	if (lseek(clone_fd, clone_offset, SEEK_SET) == (off_t)-1) {
		ret = -errno;
		fprintf(stderr, "ERROR: failed to lseek %s. %s\n",
			full_clone_path, strerror(-ret));
		goto out;
	}

	while (len > 0) {
		char buf[4096];
		uint64_t sub_len;
		int read_len;
		int write_len;
		int offset;

		sub_len = sizeof(buf);
		if (sub_len > len)
			sub_len = len;
		read_len = read(clone_fd, buf, sub_len);
		if (read_len < 0) {
			if (errno == EINTR)
				continue;
			ret = -errno;
			fprintf(stderr, "ERROR: failed to read from %s. %s\n",
				full_clone_path, strerror(-ret));
			goto out;
		} else if (read_len == 0) {
			ret = -ESPIPE; /* Hmm */
			fprintf(stderr,
				"ERROR: premature EOF while reading from %s.\n",
				full_clone_path);
			goto out;
		}

		len -= read_len;
		offset = 0;
		do {
			write_len = write(frctx->write_fd, buf + offset,
					  read_len);
			if (write_len < 0) {
				if (errno == EINTR)
					continue;
				ret = -errno;
				fprintf(stderr,
					"ERROR: failed to write to %s. %s\n",
					full_dst_path, strerror(-ret));
				goto out;
			}
			offset += write_len;
			read_len -= write_len;
		} while (read_len > 0);
	}

out:
	free(full_dst_path);
	free(full_clone_path);
	if (clone_fd != -1)
		close(clone_fd);
	return ret;
}

static int process_set_xattr(struct far_rcv_ctx *frctx, const char *path,
			     const char *name, const void *data, int len)
{
	int ret = 0;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr,
			"set_xattr %s - name=%s data_len=%d data=%.*s\n",
			path, name, len, len, (char*)data);

	ret = far_rcv_lsetxattr(full_path, name, data, len);
	if (ret < 0) {
		fprintf(stderr,
			"ERROR: far_rcv_lsetxattr %s %s=%.*s failed. %s\n",
			path, name, len, (char*)data, strerror(-ret));
		goto out;
	}

out:
	free(full_path);
	return ret;
}

static int process_remove_xattr(struct far_rcv_ctx *frctx, const char *path,
				const char *name)
{
	int ret = 0;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "remove_xattr %s - name=%s\n", path, name);

	ret = far_rcv_lremovexattr(full_path, name);
	if (ret < 0) {
		fprintf(stderr,
			"ERROR: far_rcv_lremovexattr %s %s failed. %s\n",
			path, name, strerror(-ret));
		goto out;
	}

out:
	free(full_path);
	return ret;
}

static int process_truncate(struct far_rcv_ctx *frctx, const char *path,
			    uint64_t size)
{
	int ret = 0;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "truncate %s size=%" PRIu64 "\n", path, size);

	ret = truncate(full_path, size);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: truncate %s failed. %s\n", path,
			strerror(-ret));
		goto out;
	}

out:
	free(full_path);
	return ret;
}

static int process_chmod(struct far_rcv_ctx *frctx, const char *path,
			 uint64_t mode)
{
	int ret = 0;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "chmod %s - mode=0%o\n", path, (int)mode);

	ret = chmod(full_path, mode);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: chmod %s failed. %s\n",
				path, strerror(-ret));
		goto out;
	}

out:
	free(full_path);
	return ret;
}

static int process_chown(struct far_rcv_ctx *frctx, const char *path,
			 uint64_t uid, uint64_t gid)
{
	int ret = 0;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);

	if (frctx->verbose >= 2)
		fprintf(stderr, "chown %s - uid=%" PRIu64 ", gid=%" PRIu64 "\n",
			path, uid, gid);

	ret = lchown(full_path, uid, gid);
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: chown %s failed. %s\n", path,
			strerror(-ret));
		goto out;
	}

out:
	free(full_path);
	return ret;
}

static int process_utimes(struct far_rcv_ctx *frctx, const char *path,
			  struct timespec *at, struct timespec *mt,
			  struct timespec *ct)
{
	int ret = 0;
	char *full_path = far_rcv_path_cat(frctx->current_base_path, path);
#if defined (HAVE_UTIMENSAT)
	struct timespec tv[2];

	if (frctx->verbose >= 2)
		fprintf(stderr, "utimes %s\n", path);

	tv[0] = *at;
	tv[1] = *mt;
	ret = utimensat(AT_FDCWD, full_path, tv, AT_SYMLINK_NOFOLLOW);
#else /* !defined (HAVE_UTIMENSAT) */
	struct utimbuf tv;

	tv.actime = at->tv_sec;
	tv.modtime = mt->tv_sec;
	ret = utime(path, &tv);
#endif /* !defined (HAVE_UTIMENSAT) */
	if (ret < 0) {
		ret = -errno;
		fprintf(stderr, "ERROR: utimes %s failed. %s\n", path,
			strerror(-ret));
		goto out;
	}

out:
	free(full_path);
	return ret;
}

char *far_rcv_path_cat(const char *p1, const char *p2)
{
	int p1_len = strlen(p1);
	int p2_len = strlen(p2);
	char *new = malloc(p1_len + p2_len + 2);

	if (p1_len && p1[p1_len - 1] == '/')
		p1_len--;
	if (p2_len && p2[p2_len - 1] == '/')
		p2_len--;
	sprintf(new, "%.*s/%.*s", p1_len, p1, p2_len, p2);
	return new;
}