[RFC Patch v2 39/45] block-colo: implement colo disk replication

[Wen Congyang <wency@cn.fujitsu.com>]

 TODO:
 update block-remus to use async io to instead
 of mread/mwrite.

Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
---
 tools/blktap2/drivers/Makefile            |    3 +
 tools/blktap2/drivers/block-colo.c        | 1151 +++++++++++++++++++++++++++++
 tools/blktap2/drivers/block-replication.c |  196 +++++
 tools/blktap2/drivers/block-replication.h |   56 ++
 tools/blktap2/drivers/tapdisk-disktype.c  |    9 +
 tools/blktap2/drivers/tapdisk-disktype.h  |    3 +-
 6 files changed, 1417 insertions(+), 1 deletion(-)
 create mode 100644 tools/blktap2/drivers/block-colo.c

diff --git a/tools/blktap2/drivers/Makefile b/tools/blktap2/drivers/Makefile
index c7a2ca4..0f91ccf 100644
--- a/tools/blktap2/drivers/Makefile
+++ b/tools/blktap2/drivers/Makefile
@@ -28,6 +28,8 @@ REMUS-OBJS  += hashtable.o
 REMUS-OBJS  += hashtable_itr.o
 REMUS-OBJS  += hashtable_utility.o
 
+COLO-OBJS += block-colo.o
+
 tapdisk2 tapdisk-stream tapdisk-diff $(QCOW_UTIL): AIOLIBS := -laio
 
 MEMSHRLIBS :=
@@ -74,6 +76,7 @@ BLK-OBJS-y  += aes.o
 BLK-OBJS-y  += md5.o
 BLK-OBJS-y  += $(PORTABLE-OBJS-y)
 BLK-OBJS-y  += $(REMUS-OBJS)
+BLK-OBJS-y  += $(COLO-OBJS)
 
 all: $(IBIN) lock-util qcow-util
 
diff --git a/tools/blktap2/drivers/block-colo.c b/tools/blktap2/drivers/block-colo.c
new file mode 100644
index 0000000..565a386
--- /dev/null
+++ b/tools/blktap2/drivers/block-colo.c
@@ -0,0 +1,1151 @@
+/*
+ * Copyright (C) 2014 FUJITSU LIMITED
+ * Author: Wen Congyang <wency@cn.fujitsu.com>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published
+ * by the Free Software Foundation; version 2.1 only. with the special
+ * exception on linking described in file LICENSE.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ */
+
+#include "tapdisk.h"
+#include "tapdisk-server.h"
+#include "tapdisk-driver.h"
+#include "tapdisk-interface.h"
+#include "block-replication.h"
+
+#include <errno.h>
+#include <stdlib.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <sys/un.h>
+#include <unistd.h>
+
+/* connect retry timeout (seconds) */
+#define COLO_CONNRETRY_TIMEOUT  1
+
+/* timeout for reads and writes in second */
+#define HEARTBEAT_S 1
+
+/* TAPDISK_DATA_REQUESTS I/O requests + commit flag */
+#define MAX_COLO_REQUEST        TAPDISK_DATA_REQUESTS + 1
+
+#undef DPRINTF
+#undef EPRINTF
+#define DPRINTF(_f, _a...) syslog (LOG_DEBUG, "COLO: " _f, ## _a)
+#define EPRINTF(_f, _a...) syslog (LOG_ERR, "COLO: " _f, ## _a)
+
+#define TDCOLO_WRITE "wreq"
+#define TDCOLO_COMMIT "creq"
+#define TDCOLO_DONE "done"
+#define TDCOLO_FAIL "fail"
+
+enum tdcolo_mode {
+	mode_invalid = 0,
+	mode_unprotected,
+	mode_primary,
+	mode_backup,
+
+	/*
+	 * If we find some internal error in backup mode, we cannot
+	 * switch to unprotected mode.
+	 */
+	mode_failed,
+};
+
+enum {
+	colo_io,
+	colo_commit,
+};
+
+typedef struct queued_io {
+	int type;
+	union {
+		td_request_t treq;
+		char *buff; /* TDCOLO_COMMIT */
+	};
+} queued_io_t;
+
+struct queued_io_ring {
+	/* waste one slot to distinguish between empty and full */
+	queued_io_t qio[MAX_COLO_REQUEST + 1];
+	unsigned int prod;
+	unsigned int cons;
+};
+
+typedef struct colo_control {
+	/*
+	 * socket file, the user writes "flush" to this socket, and then
+	 * we write the result to it.
+	 */
+	char *path;
+	int listen_fd;
+	event_id_t listen_id;
+
+	int io_fd;
+	event_id_t io_id;
+} colo_control_t;
+
+struct tdcolo_state {
+	colo_control_t ctl;
+
+	/* async connection */
+	td_replication_connect_t t;
+	/* replication channel */
+	td_async_io_t rio, wio;
+
+	/*
+	 * queue I/O requests, and they will be forwarded to backup
+	 * asynchronously.
+	 */
+	struct queued_io_ring qio_ring;
+
+	/* ramdisk data */
+	struct ramdisk ramdisk;
+	/*
+	 * The primary write request is queued in this
+	 * hashtable, and will be flushed to ramdisk when
+	 * the checkpoint finishes.
+	 */
+	struct hashtable *h;
+	/*
+	 * The secondary vm write request is queued in this
+	 * hashtable, and will be dropped when the checkpoint
+	 * finishes or flushed to ramdisk after failover.
+	 */
+	struct hashtable *local;
+
+	/* mode methods */
+	enum tdcolo_mode mode;
+	/* It will be called when switching mode */
+	int (*queue_flush)(struct tdcolo_state *c);
+
+	char request[5];
+	char header[sizeof(uint32_t) + sizeof(uint64_t)];
+	int commit;
+	void *buff;
+	int bsize;
+	int sector_size;
+};
+
+struct tap_disk tapdisk_colo;
+
+static void colo_control_respond(colo_control_t *ctl, const char *response);
+static int switch_mode(struct tdcolo_state *c, enum tdcolo_mode mode);
+
+/* ======== common functions ======== */
+static int check_read_result(td_async_io_t *rio, int realsize,
+			     const char *target)
+{
+	if (realsize < 0) {
+		/* internal error */
+		EPRINTF("error reading from %s\n", target);
+		return ERROR_INTERNAL;
+	} else if (realsize < rio->size) {
+		/* timeout or I/O error */
+		EPRINTF("error reading from %s\n", target);
+		return ERROR_IO;
+	}
+
+	return 0;
+}
+
+static int check_write_result(td_async_io_t *wio, int realsize,
+			      const char * target)
+{
+	if (realsize < 0) {
+		/* internal error */
+		EPRINTF("error writing to %s\n", target);
+		return ERROR_INTERNAL;
+	} else if (realsize == 0) {
+		/* timeout or I/O error */
+		EPRINTF("error writing to %s\n", target);
+		return ERROR_IO;
+	}
+
+	return 0;
+}
+
+/* ======= ring functions ======== */
+static inline unsigned int ring_next(unsigned int pos)
+{
+	if (++pos > MAX_COLO_REQUEST)
+		return 0;
+
+	return pos;
+}
+
+static inline int ring_isempty(struct queued_io_ring* ring)
+{
+	return ring->cons == ring->prod;
+}
+
+static inline int ring_isfull(struct queued_io_ring* ring)
+{
+	return ring_next(ring->prod) == ring->cons;
+}
+
+static void ring_add_request(struct queued_io_ring *ring,
+			     const td_request_t *treq)
+{
+	/* If ring is full, it means that tapdisk2 has some bug */
+	if (ring_isfull(ring)) {
+		EPRINTF("OOPS, ring is full\n");
+		exit(1);
+	}
+
+	ring->qio[ring->prod].type = colo_io;
+	ring->qio[ring->prod].treq = *treq;
+	ring->prod = ring_next(ring->prod);
+}
+
+static void ring_add_commit_flag(struct queued_io_ring *ring)
+{
+	/* If ring is full, it means that tapdisk2 has some bug */
+	if (ring_isfull(ring)) {
+		EPRINTF("OOPS, ring is full\n");
+		exit(1);
+	}
+
+	ring->qio[ring->prod].type = colo_commit;
+	ring->qio[ring->prod].buff = TDCOLO_COMMIT;
+	ring->prod = ring_next(ring->prod);
+}
+
+/* return the first queued I/O request */
+static queued_io_t *ring_peek(struct queued_io_ring *ring)
+{
+	queued_io_t *qio;
+
+	if (ring_isempty(ring))
+		return NULL;
+
+	qio = &ring->qio[ring->cons];
+	return qio;
+}
+
+/* consume the first queued I/O request, and return it */
+static queued_io_t *ring_get(struct queued_io_ring *ring)
+{
+	queued_io_t *qio;
+
+	if (ring_isempty(ring))
+		return NULL;
+
+	qio = &ring->qio[ring->cons];
+	ring->cons = ring_next(ring->cons);
+	return qio;
+}
+
+/* ======== primary read/write functions ======== */
+static void primary_write_header(td_async_io_t *wio, int realsize, int errnoval);
+static void primary_write_data(td_async_io_t *wio, int realsize, int errnoval);
+static void primary_forward_done(td_async_io_t *wio, int realsize, int errnoval);
+static void primary_read_done(td_async_io_t *rio, int realsize, int errnoval);
+
+/*
+ * It is called when we cannot connect to backup, or find I/O error when
+ * reading/writing.
+ */
+static void primary_failed(struct tdcolo_state *c, int rc)
+{
+	td_replication_connect_kill(&c->t);
+	td_async_io_kill(&c->rio);
+	td_async_io_kill(&c->wio);
+	if (rc == ERROR_INTERNAL)
+		EPRINTF("switch to unprotected mode due to internal error");
+	if (rc == ERROR_CLOSE)
+		DPRINTF("switch to unprotected mode before closing");
+	switch_mode(c, mode_unprotected);
+}
+
+static void primary_waio(struct tdcolo_state *c, void *buff, size_t size,
+			 taio_callback *callback)
+{
+	td_async_io_t *wio = &c->wio;
+
+	wio->fd = c->t.fd;
+	wio->timeout_s = HEARTBEAT_S;
+	wio->mode = td_async_write;
+	wio->buff = buff;
+	wio->size = size;
+	wio->callback = callback;
+
+	if (td_async_io_start(wio))
+		primary_failed(c, ERROR_INTERNAL);
+}
+
+static void primary_raio(struct tdcolo_state *c)
+{
+	td_async_io_t *rio = &c->rio;
+
+	if (c->t.fd < 0)
+		return;
+
+	rio->fd = c->t.fd;
+	rio->timeout_s = 0;
+	rio->mode = td_async_read;
+	rio->buff = c->request;
+	rio->size = sizeof(c->request) - 1;
+	rio->callback = primary_read_done;
+
+	if (td_async_io_start(rio))
+		primary_failed(c, ERROR_INTERNAL);
+}
+
+static void primary_handle_queued_io(struct tdcolo_state *c)
+{
+	struct queued_io_ring *qring = &c->qio_ring;
+	unsigned int cons;
+	queued_io_t *qio;
+	int rc;
+
+	while (!ring_isempty(qring)) {
+		qio = ring_peek(qring);
+		if (qio->type == colo_commit) {
+			primary_waio(c, qio->buff, strlen(qio->buff),
+				     primary_forward_done);
+			return;
+		}
+
+		if (qio->treq.op == TD_OP_WRITE) {
+			primary_waio(c, TDCOLO_WRITE, strlen(TDCOLO_WRITE),
+				     primary_write_header);
+			return;
+		}
+
+		td_forward_request(qio->treq);
+		ring_get(qring);
+	}
+}
+
+/* wait for "done" message to commit checkpoint */
+static void primary_read_done(td_async_io_t *rio, int realsize, int errnoval)
+{
+	struct tdcolo_state *c = CONTAINER_OF(rio, *c, rio);
+	char *req = c->request;
+	int rc;
+
+	rc = check_read_result(rio, realsize, "backup");
+	if (rc)
+		goto err;
+
+	rc = ERROR_INTERNAL;
+	req[4] = '\0';
+
+	if (c->commit != 1) {
+		EPRINTF("received unexpected message: %s\n", req);
+		goto err;
+	}
+
+	c->commit--;
+
+	if (strcmp(req, TDCOLO_DONE)) {
+		EPRINTF("received unknown message: %s\n", req);
+		goto err;
+	}
+
+	/* checkpoint committed, inform msg_fd */
+	colo_control_respond(&c->ctl, TDCOLO_DONE);
+	primary_raio(c);
+
+	return;
+err:
+	colo_control_respond(&c->ctl, TDCOLO_FAIL);
+	primary_failed(c, rc);
+}
+
+static void primary_write_header(td_async_io_t *wio, int realsize, int errnoval)
+{
+	struct tdcolo_state *c = CONTAINER_OF(wio, *c, wio);
+	queued_io_t *qio = ring_peek(&c->qio_ring);
+	uint32_t *sectors = (uint32_t *)c->header;
+	uint64_t *sector = (uint64_t *)(c->header + sizeof(uint32_t));
+	int rc;
+
+	rc = check_write_result(wio, realsize, "backup");
+	if (rc) {
+		primary_failed(c, rc);
+		return;
+	}
+
+	*sectors = qio->treq.secs;
+	*sector = qio->treq.sec;
+
+	primary_waio(c, c->header, sizeof(c->header), primary_write_data);
+}
+
+static void primary_write_data(td_async_io_t *wio, int realsize, int errnoval)
+{
+	struct tdcolo_state *c = CONTAINER_OF(wio, *c, wio);
+	queued_io_t *qio = ring_peek(&c->qio_ring);
+	int rc;
+
+	rc = check_write_result(wio, realsize, "backup");
+	if (rc) {
+		primary_failed(c, rc);
+		return;
+	}
+
+	primary_waio(c, qio->treq.buf, qio->treq.secs * c->sector_size,
+		     primary_forward_done);
+}
+
+static void primary_forward_done(td_async_io_t *wio, int realsize, int errnoval)
+{
+	struct tdcolo_state *c = CONTAINER_OF(wio, *c, wio);
+	queued_io_t *qio;
+	struct td_request_t *treq;
+	int rc;
+
+	rc = check_write_result(wio, realsize, "backup");
+	if (rc) {
+		primary_failed(c, rc);
+		return;
+	}
+
+	qio = ring_get(&c->qio_ring);
+	if (qio->type == colo_io)
+		td_forward_request(qio->treq);
+	else
+		c->commit--;
+
+	primary_handle_queued_io(c);
+}
+
+static void primary_queue_read(td_driver_t *driver, td_request_t treq)
+{
+	struct tdcolo_state *c = driver->data;
+	struct queued_io_ring *ring = &c->qio_ring;
+
+	if (ring_isempty(ring)) {
+		/* just pass read through */
+		td_forward_request(treq);
+		return;
+	}
+
+	ring_add_request(ring, &treq);
+	if (td_replication_connect_status(&c->t) != 1)
+		return;
+
+	if (!td_async_io_is_running(&c->wio))
+		primary_handle_queued_io(c);
+}
+
+static void primary_queue_write(td_driver_t *driver, td_request_t treq)
+{
+	struct tdcolo_state *c = driver->data;
+	struct queued_io_ring *ring = &c->qio_ring;
+
+	ring_add_request(ring, &treq);
+	if (td_replication_connect_status(&c->t) != 1)
+		return;
+
+	if (!td_async_io_is_running(&c->wio))
+		primary_handle_queued_io(c);
+}
+
+/* It is called when the user write "flush" to control file. */
+static int client_flush(struct tdcolo_state *c)
+{
+	if (td_replication_connect_status(&c->t) != 1)
+		return 0;
+
+	if (c->commit > 0) {
+		EPRINTF("the last commit is not finished\n");
+		colo_control_respond(&c->ctl, TDCOLO_FAIL);
+		primary_failed(c, ERROR_INTERNAL);
+		return -1;
+	}
+
+	ring_add_commit_flag(&c->qio_ring);
+	c->commit = 2;
+	if (!td_async_io_is_running(&c->wio))
+		primary_handle_queued_io(c);
+
+	return 0;
+}
+
+/* It is called when switching the mode from primary to unprotected */
+static int primary_flush(struct tdcolo_state *c)
+{
+	struct queued_io_ring *qring = &c->qio_ring;
+	queued_io_t *qio;
+
+	if (ring_isempty(qring))
+		return 0;
+
+	while (!ring_isempty(qring)) {
+		qio = ring_get(qring);
+
+		if (qio->type == colo_commit) {
+			colo_control_respond(&c->ctl, TDCOLO_FAIL);
+			c->commit = 0;
+			continue;
+		}
+
+		td_forward_request(qio->treq);
+	}
+
+	return 0;
+}
+
+static void colo_client_established(td_replication_connect_t *t, int rc)
+{
+	struct tdcolo_state *c = CONTAINER_OF(t, *c, t);
+
+	if (rc) {
+		primary_failed(c, rc);
+		return;
+	}
+
+	/* the connect succeeded and handle the queued requests */
+	primary_handle_queued_io(c);
+
+	primary_raio(c);
+}
+
+static int primary_start(struct tdcolo_state *c)
+{
+	DPRINTF("activating client mode\n");
+
+	tapdisk_colo.td_queue_read = primary_queue_read;
+	tapdisk_colo.td_queue_write = primary_queue_write;
+	c->queue_flush = primary_flush;
+
+	c->t.callback = colo_client_established;
+	return td_replication_client_start(&c->t);
+}
+
+/* ======== backup read/write functions ======== */
+static void backup_read_header_done(td_async_io_t *rio, int realsize,
+				    int errnoval);
+static void backup_read_data_done(td_async_io_t *rio, int realsize,
+				  int errnoval);
+static void backup_write_done(td_async_io_t *wio, int realsize, int errnoval);
+
+static void backup_failed(struct tdcolo_state *c, int rc)
+{
+	td_replication_connect_kill(&c->t);
+	td_async_io_kill(&c->rio);
+	td_async_io_kill(&c->wio);
+
+	if (rc == ERROR_INTERNAL) {
+		EPRINTF("switch to failed mode due to internal error");
+		switch_mode(c, mode_failed);
+		return;
+	}
+
+	if (rc == ERROR_CLOSE)
+		DPRINTF("switch to unprotected mode before closing");
+
+	switch_mode(c, mode_unprotected);
+}
+
+static void backup_raio(struct tdcolo_state *c, void *buff, int size,
+			int timeout_s, taio_callback *callback)
+{
+	td_async_io_t *rio = &c->rio;
+
+	rio->fd = c->t.fd;
+	rio->timeout_s = timeout_s;
+	rio->mode = td_async_read;
+	rio->buff = buff;
+	rio->size = size;
+	rio->callback = callback;
+
+	if (td_async_io_start(rio)) {
+		EPRINTF("cannot start read aio\n");
+		backup_failed(c, ERROR_INTERNAL);
+	}
+}
+
+static void backup_waio(struct tdcolo_state *c)
+{
+	td_async_io_t *wio = &c->wio;
+
+	wio->fd = c->t.fd;
+	wio->timeout_s = HEARTBEAT_S;
+	wio->mode = td_async_write;
+	wio->buff = TDCOLO_DONE;
+	wio->size = strlen(TDCOLO_DONE);
+	wio->callback = backup_write_done;
+
+	if (td_async_io_start(wio)) {
+		EPRINTF("cannot start write aio\n");
+		backup_failed(c, ERROR_INTERNAL);
+	}
+}
+
+static void backup_read_req_done(td_async_io_t *rio, int realsize,
+				 int errnoval)
+{
+	struct tdcolo_state *c = CONTAINER_OF(rio, *c, rio);
+	char *req = c->request;
+	int rc;
+
+	rc = check_read_result(rio, realsize, "primary");
+	if (rc)
+		goto err;
+
+	rc = ERROR_INTERNAL;
+	req[4] = '\0';
+
+	if (!strcmp(req, TDCOLO_WRITE)) {
+		backup_raio(c, c->header, sizeof(c->header), HEARTBEAT_S,
+			    backup_read_header_done);
+		return;
+	} else if (!strcmp(req, TDCOLO_COMMIT)) {
+		ramdisk_destroy_hashtable(c->local);
+		c->local = ramdisk_new_hashtable();
+		if (!c->local) {
+			EPRINTF("error creating local hashtable\n");
+			goto err;
+		}
+		rc = ramdisk_start_flush(&c->ramdisk, &c->h);
+		if (rc) {
+			EPRINTF("error flushing queued I/O\n");
+			goto err;
+		}
+
+		backup_waio(c);
+	} else {
+		EPRINTF("unsupported request: %s\n", req);
+		goto err;
+	}
+
+	return;
+
+err:
+	backup_failed(c, ERROR_INTERNAL);
+	return;
+}
+
+static void backup_read_header_done(td_async_io_t *rio, int realsize,
+				    int errnoval)
+{
+	struct tdcolo_state *c = CONTAINER_OF(rio, *c, rio);
+	uint32_t *sectors = (uint32_t *)c->header;
+	int rc;
+
+	rc = check_read_result(rio, realsize, "primary");
+	if (rc)
+		goto err;
+
+	rc = ERROR_INTERNAL;
+	if (*sectors * c->sector_size > c->bsize) {
+		EPRINTF("write request is too large: %d/%d\n",
+			*sectors * c->sector_size, c->bsize);
+		goto err;
+	}
+
+	backup_raio(c, c->buff, *sectors * c->sector_size, HEARTBEAT_S,
+		    backup_read_data_done);
+
+	return;
+err:
+	backup_failed(c, rc);
+}
+
+static void backup_read_data_done(td_async_io_t *rio, int realsize,
+				  int errnoval)
+{
+	struct tdcolo_state *c = CONTAINER_OF(rio, *c, rio);
+	uint32_t *sectors = (uint32_t *)c->header;
+	uint64_t *sector = (uint64_t *)(c->header + sizeof(uint32_t));
+	int rc;
+
+	rc = check_read_result(rio, realsize, "primary");
+	if (rc)
+		goto err;
+
+	rc = ramdisk_write_to_hashtable(c->h, *sector, *sectors,
+					c->sector_size, c->buff, "COLO");
+	if (rc) {
+		EPRINTF("cannot write primary data to hashtable\n");
+		rc = ERROR_INTERNAL;
+		goto err;
+	}
+
+	backup_raio(c, c->request, sizeof(c->request) - 1, 0,
+		    backup_read_req_done);
+
+	return;
+err:
+	backup_failed(c, rc);
+}
+
+static void backup_write_done(td_async_io_t *wio, int realsize, int errnoval)
+{
+	struct tdcolo_state *c = CONTAINER_OF(wio, *c, wio);
+	int rc;
+
+	rc = check_write_result(wio, realsize, "primary");
+	if (rc) {
+		backup_failed(c, rc);
+		return;
+	}
+
+	backup_raio(c, c->request, sizeof(c->request) - 1, 0,
+		    backup_read_req_done);
+}
+
+static void colo_server_established(td_replication_connect_t *t, int rc)
+{
+	struct tdcolo_state *c = CONTAINER_OF(t, *c, t);
+
+	if (rc) {
+		backup_failed(c, rc);
+		return;
+	}
+
+	backup_raio(c, c->request, sizeof(c->request) - 1, 0,
+		    backup_read_req_done);
+}
+
+/* It is called when switching the mode from backup to unprotected */
+static int backup_flush(struct tdcolo_state *c)
+{
+	int rc;
+
+	rc = ramdisk_start_flush(&c->ramdisk, &c->local);
+	if (rc)
+		EPRINTF("error flushing local queued I/O\n");
+
+	return 0;
+}
+
+static void backup_queue_read(td_driver_t *driver, td_request_t treq)
+{
+	struct tdcolo_state *c = driver->data;
+
+	if (ramdisk_read_from_hashtable(c->local, treq.sec, treq.secs,
+					c->sector_size, treq.buf))
+		/* FIXME */
+		td_forward_request(treq);
+	else
+		/* complete the request */
+		td_complete_request(treq, 0);
+}
+
+static void backup_queue_write(td_driver_t *driver, td_request_t treq)
+{
+	struct tdcolo_state *c = driver->data;
+	int rc;
+
+	rc = ramdisk_write_to_hashtable(c->local, treq.sec, treq.secs,
+					c->sector_size, treq.buf,
+					"COLO");
+	if (rc)
+		td_complete_request(treq, -EBUSY);
+	else
+		td_complete_request(treq, 0);
+}
+
+static int backup_start(struct tdcolo_state *c)
+{
+	tapdisk_colo.td_queue_read = backup_queue_read;
+	tapdisk_colo.td_queue_write = backup_queue_write;
+	c->queue_flush = backup_flush;
+
+	c->h = ramdisk_new_hashtable();
+	c->local = ramdisk_new_hashtable();
+	if (!c->h || !c->local)
+		return -1;
+
+	c->bsize = sysconf(_SC_PAGESIZE);
+	c->buff = malloc(c->bsize);
+	if (!c->buff)
+		return -1;
+
+	return 0;
+}
+
+/* ======== unprotected read/write functions ======== */
+void unprotected_queue_io(td_driver_t *driver, td_request_t treq)
+{
+	struct tdcolo_state *c = driver->data;
+
+	/* wait for previous ramdisk to flush  before servicing I/O */
+	if (ramdisk_writes_inflight(&c->ramdisk)) {
+		ramdisk_flush(&c->ramdisk);
+		td_complete_request(treq, -EBUSY);
+	} else {
+		/* here we just pass I/O through */
+		td_forward_request(treq);
+	}
+}
+
+static int unprotected_start(struct tdcolo_state *c)
+{
+	DPRINTF("failure detected, activating passthrough\n");
+
+	/* install the unprotected read/write handlers */
+	tapdisk_colo.td_queue_read = unprotected_queue_io;
+	tapdisk_colo.td_queue_write = unprotected_queue_io;
+	c->queue_flush = NULL;
+
+	return 0;
+}
+
+/* ======== failed read/write functions ======== */
+static void failed_queue_io(td_driver_t *driver, td_request_t treq)
+{
+	td_complete_request(treq, -EIO);
+}
+
+static int failed_start(struct tdcolo_state *c)
+{
+	tapdisk_colo.td_queue_read = failed_queue_io;
+	tapdisk_colo.td_queue_write = failed_queue_io;
+	c->queue_flush = NULL;
+
+	return 0;
+}
+
+/* ======== control ======== */
+static void colo_control_accept(event_id_t id, char mode, void *private);
+static void colo_control_handle_request(event_id_t id, char mode,
+					void *private);
+static void colo_control_close(colo_control_t *ctl);
+
+static void colo_control_init(colo_control_t *ctl)
+{
+	ctl->listen_fd = -1;
+	ctl->listen_id = -1;
+	ctl->io_fd = -1;
+	ctl->io_id = -1;
+}
+
+static int colo_create_control_socket(colo_control_t *ctl, const char *name)
+{
+	int i, l;
+	struct sockaddr_un saddr;
+	event_id_t id;
+	int rc;
+
+	/* first we must ensure that BLKTAP_CTRL_DIR exists */
+	if (mkdir(BLKTAP_CTRL_DIR, 0755) && errno != EEXIST) {
+		rc = -errno;
+		EPRINTF("error creating directory %s: %d\n",
+			BLKTAP_CTRL_DIR, errno);
+		goto fail;
+	}
+
+	/* use the device name to create the control socket path */
+	if (asprintf(&ctl->path, BLKTAP_CTRL_DIR "/colo_%s", name) < 0) {
+		rc = -errno;
+		goto fail;
+	}
+
+	/* scrub socket pathname  */
+	l = strlen(ctl->path);
+	for (i = strlen(BLKTAP_CTRL_DIR) + 1; i < l; i++) {
+		if (strchr(":/", ctl->path[i]))
+			ctl->path[i] = '_';
+	}
+
+	if (unlink(ctl->path) && errno != ENOENT) {
+		rc = -errno;
+		EPRINTF("failed to unlink %s: %d\n", ctl->path, errno);
+		goto fail;
+	}
+
+	ctl->listen_fd = socket(AF_UNIX, SOCK_STREAM, 0);
+	if (ctl->listen_fd == -1) {
+		rc = -errno;
+		EPRINTF("failed to create control socket: %d\n", errno);
+		goto fail;
+	}
+
+	memset(&saddr, 0, sizeof(saddr));
+	strncpy(saddr.sun_path, ctl->path, sizeof(saddr.sun_path));
+	saddr.sun_family = AF_UNIX;
+
+	rc = bind(ctl->listen_fd, (const struct sockaddr *)&saddr,
+		  sizeof(saddr));
+	if (rc == -1) {
+		rc = -errno;
+		EPRINTF("failed to bind to %s: %d\n", saddr.sun_path, errno);
+		goto fail;
+	}
+
+	rc = listen(ctl->listen_fd, 10);
+	if (rc == -1) {
+		rc = -errno;
+		EPRINTF("failed to listen: %d\n", errno);
+		goto fail;
+	}
+
+	id = tapdisk_server_register_event(SCHEDULER_POLL_READ_FD,
+					   ctl->listen_fd, 0,
+					   colo_control_accept, ctl);
+	if (id < 0) {
+		EPRINTF("failed to add watch: %d\n", id);
+		rc = id;
+		goto fail;
+	}
+
+	ctl->listen_id = id;
+	return 0;
+
+fail:
+	colo_control_close(ctl);
+	return rc;
+}
+
+static void colo_control_accept(event_id_t id, char mode, void *private)
+{
+	colo_control_t *ctl = private;
+	int fd;
+
+	fd = accept(ctl->listen_fd, NULL, NULL);
+	if (fd == -1) {
+		EPRINTF("failed to accept new control connection: %d\n", errno);
+		return;
+	}
+
+	if (ctl->io_fd >= 0) {
+		EPRINTF("cannot accept two control connections\n");
+		close(fd);
+		return;
+	}
+
+	id = tapdisk_server_register_event(SCHEDULER_POLL_READ_FD,
+					   fd, 0,
+					   colo_control_handle_request,
+					   ctl);
+	if (id < 0) {
+		close(fd);
+		EPRINTF("failed to register new control event: %d\n", id);
+		return;
+	}
+
+	ctl->io_fd = fd;
+	ctl->io_id = id;
+}
+
+static void colo_control_handle_request(event_id_t id, char mode, void *private)
+{
+	colo_control_t *ctl = private;
+	struct tdcolo_state *c = CONTAINER_OF(ctl, *c, ctl);
+	char req[6];
+	int rc;
+
+	rc = read(ctl->io_fd, req, sizeof(req) - 1);
+	if (!rc) {
+		EPRINTF("0-byte read received, close control socket\n");
+		goto err;
+	}
+
+	if (rc < 0) {
+		EPRINTF("error reading from control socket: %d\n", errno);
+		goto err;
+	}
+
+	req[rc] = '\0';
+	if (strncmp(req, "flush", 5)) {
+		EPRINTF("unknown command: %s\n", req);
+		colo_control_respond(ctl, TDCOLO_FAIL);
+		return;
+	}
+
+	if (c->mode != mode_primary) {
+		EPRINTF("invalid mode: %d\n", c->mode);
+		colo_control_respond(ctl, TDCOLO_FAIL);
+		return;
+	}
+
+	client_flush(c);
+	return;
+
+err:
+	UNREGISTER_EVENT(ctl->io_id);
+	CLOSE_FD(ctl->io_fd);
+	return;
+}
+
+static void colo_control_respond(colo_control_t *ctl, const char *response)
+{
+	int rc;
+
+	if (ctl->io_fd < 0)
+		return;
+
+	rc = write(ctl->io_fd, response, strlen(response));
+	if (rc < 0) {
+		EPRINTF("error writing notification: %d\n", errno);
+		CLOSE_FD(ctl->io_fd);
+	}
+}
+
+static void colo_control_close(colo_control_t *ctl)
+{
+	UNREGISTER_EVENT(ctl->listen_id);
+	UNREGISTER_EVENT(ctl->io_id);
+	CLOSE_FD(ctl->listen_fd);
+	CLOSE_FD(ctl->io_fd);
+
+	if (ctl->path) {
+		unlink(ctl->path);
+		free(ctl->path);
+		ctl->path = NULL;
+	}
+}
+
+/* ======== interface ======== */
+static int tdcolo_close(td_driver_t *driver);
+
+static int switch_mode(struct tdcolo_state *c, enum tdcolo_mode mode)
+{
+	int rc;
+
+	if (mode == c->mode)
+		return 0;
+
+	if (c->queue_flush)
+		if ((rc = c->queue_flush(c)) < 0) {
+			/* fall back to unprotected mode on error */
+			EPRINTF("switch_mode: error flushing queue (old: %d, new: %d)",
+				c->mode, mode);
+			mode = mode_unprotected;
+		}
+
+	if (mode == mode_unprotected)
+		rc = unprotected_start(c);
+	else if (mode == mode_primary)
+		rc = primary_start(c);
+	else if (mode == mode_backup)
+		rc = backup_start(c);
+	else if (mode == mode_failed)
+		rc = failed_start(c);
+	else {
+		EPRINTF("unknown mode requested: %d\n", mode);
+		rc = -1;
+	}
+
+	if (!rc)
+		c->mode = mode;
+
+	return rc;
+}
+
+static int tdcolo_open(td_driver_t *driver, td_image_t *image, td_uuid_t uuid)
+{
+	struct tdcolo_state *c = driver->data;
+	td_replication_connect_t *t = &c->t;
+	colo_control_t *ctl = &c->ctl;
+	ramdisk_t *ramdisk = &c->ramdisk;
+	int rc;
+	const char *name = image->name;
+	td_flag_t flags = image->flags;
+
+	DPRINTF("opening %s\n", name);
+
+	memset(c, 0, sizeof(*c));
+
+	/* init ramdisk */
+	ramdisk->log_prefix = "COLO";
+	ramdisk->sector_size = driver->info.sector_size;
+	ramdisk->image = image;
+	ramdisk_init(&c->ramdisk);
+
+	/* init async I/O */
+	td_async_io_init(&c->rio);
+	td_async_io_init(&c->wio);
+
+	c->sector_size = driver->info.sector_size;
+
+	/* init control socket */
+	colo_control_init(ctl);
+	rc = colo_create_control_socket(ctl, name);
+	if (rc)
+		return rc;
+
+	/* init async connection */
+	t->log_prefix = "COLO";
+	t->retry_timeout_s = COLO_CONNRETRY_TIMEOUT;
+	t->max_connections = 1;
+	t->callback = colo_server_established;
+	rc = td_replication_connect_init(t, name);
+	if (rc) {
+		colo_control_close(ctl);
+		return rc;
+	}
+
+	rc = td_replication_server_start(t);
+	if (!rc)
+		rc = switch_mode(c, mode_backup);
+	else if (rc == -2)
+		rc = switch_mode(c, mode_primary);
+
+	if (!rc)
+		return 0;
+
+	tdcolo_close(driver);
+	return -EIO;
+}
+
+static int tdcolo_pre_close(td_driver_t *driver)
+{
+	struct tdcolo_state *c = driver->data;
+
+	if (c->mode != mode_primary)
+		return 0;
+
+	if (td_replication_connect_status(&c->t))
+		return 0;
+
+	/*
+	 * The connection is in progress, and we may queue some
+	 * I/O requests.
+	 */
+	primary_failed(c, ERROR_CLOSE);
+	return 0;
+}
+
+static int tdcolo_close(td_driver_t *driver)
+{
+	struct tdcolo_state *c = driver->data;
+
+	DPRINTF("closing\n");
+	ramdisk_destroy(&c->ramdisk);
+	ramdisk_destroy_hashtable(c->h);
+	ramdisk_destroy_hashtable(c->local);
+	td_replication_connect_kill(&c->t);
+	td_async_io_kill(&c->rio);
+	td_async_io_kill(&c->wio);
+	colo_control_close(&c->ctl);
+	free(c->buff);
+
+	return 0;
+}
+
+static int tdcolo_get_parent_id(td_driver_t *driver, td_disk_id_t *id)
+{
+	/* we shouldn't have a parent... for now */
+	return -EINVAL;
+}
+
+static int tdcolo_validate_parent(td_driver_t *driver,
+				  td_driver_t *pdriver, td_flag_t flags)
+{
+	return 0;
+}
+
+struct tap_disk tapdisk_colo = {
+	.disk_type          = "tapdisk_colo",
+	.private_data_size  = sizeof(struct tdcolo_state),
+	.td_open            = tdcolo_open,
+	.td_queue_read      = unprotected_queue_io,
+	.td_queue_write     = unprotected_queue_io,
+	.td_pre_close       = tdcolo_pre_close,
+	.td_close           = tdcolo_close,
+	.td_get_parent_id   = tdcolo_get_parent_id,
+	.td_validate_parent = tdcolo_validate_parent,
+};
diff --git a/tools/blktap2/drivers/block-replication.c b/tools/blktap2/drivers/block-replication.c
index 30eba8f..0992d19 100644
--- a/tools/blktap2/drivers/block-replication.c
+++ b/tools/blktap2/drivers/block-replication.c
@@ -911,6 +911,25 @@ int ramdisk_write_to_hashtable(struct hashtable *h, uint64_t sector,
 	return 0;
 }
 
+int ramdisk_read_from_hashtable(struct hashtable *h, uint64_t sector,
+				int nb_sectors, int sector_size,
+				char *buf)
+{
+	int i;
+	uint64_t key;
+	char *v;
+
+	for (i = 0; i < nb_sectors; i++) {
+		key = sector + i;
+		v = hashtable_search(h, &key);
+		if (!v)
+			return -1;
+		memcpy(buf + i * sector_size, v, sector_size);
+	}
+
+	return 0;
+}
+
 void ramdisk_destroy_hashtable(struct hashtable *h)
 {
 	if (!h)
@@ -918,3 +937,180 @@ void ramdisk_destroy_hashtable(struct hashtable *h)
 
 	hashtable_destroy(h, 1);
 }
+
+/* async I/O */
+static void td_async_io_readable(event_id_t id, char mode, void *private);
+static void td_async_io_writeable(event_id_t id, char mode, void *private);
+static void td_async_io_timeout(event_id_t id, char mode, void *private);
+
+void td_async_io_init(td_async_io_t *taio)
+{
+	memset(taio, 0, sizeof(*taio));
+	taio->fd = -1;
+	taio->timeout_id = -1;
+	taio->io_id = -1;
+}
+
+int td_async_io_start(td_async_io_t *taio)
+{
+	event_id_t id;
+
+	if (taio->running)
+		return -1;
+
+	if (taio->size <= 0 || taio->fd < 0)
+		return -1;
+
+	taio->running = 1;
+
+	if (taio->mode == td_async_read)
+		id = tapdisk_server_register_event(SCHEDULER_POLL_READ_FD,
+						   taio->fd, 0,
+						   td_async_io_readable,
+						   taio);
+	else if (taio->mode == td_async_write)
+		id = tapdisk_server_register_event(SCHEDULER_POLL_WRITE_FD,
+						   taio->fd, 0,
+						   td_async_io_writeable,
+						   taio);
+	else
+		id = -1;
+	if (id < 0)
+		goto err;
+	taio->io_id = id;
+
+	if (taio->timeout_s) {
+		id = tapdisk_server_register_event(SCHEDULER_POLL_TIMEOUT,
+						   -1, taio->timeout_s,
+						   td_async_io_timeout, taio);
+		if (id < 0)
+			goto err;
+		taio->timeout_id = id;
+	}
+
+	taio->used = 0;
+	return 0;
+
+err:
+	td_async_io_kill(taio);
+	return -1;
+}
+
+static void td_async_io_callback(td_async_io_t *taio, int realsize,
+				 int errnoval)
+{
+	td_async_io_kill(taio);
+	taio->callback(taio, realsize, errnoval);
+}
+
+static void td_async_io_update_timeout(td_async_io_t *taio)
+{
+	event_id_t id;
+
+	if (!taio->timeout_s)
+		return;
+
+	tapdisk_server_unregister_event(taio->timeout_id);
+	taio->timeout_id = -1;
+
+	id = tapdisk_server_register_event(SCHEDULER_POLL_TIMEOUT,
+					   -1, taio->timeout_s,
+					   td_async_io_timeout, taio);
+	if (id < 0)
+		td_async_io_callback(taio, -1, id);
+	else
+		taio->timeout_id = id;
+}
+
+static void td_async_io_readable(event_id_t id, char mode, void *private)
+{
+	td_async_io_t *taio = private;
+	int rc;
+
+	while (1) {
+		rc = read(taio->fd, taio->buff + taio->used,
+			  taio->size - taio->used);
+		if (rc < 0) {
+			if (errno == EINTR)
+				continue;
+			if (errno == EWOULDBLOCK || errno == EAGAIN)
+				break;
+
+			td_async_io_callback(taio, 0, errno);
+			return;
+		}
+
+		if (rc == 0) {
+			td_async_io_callback(taio, taio->used, 0);
+			return;
+		}
+
+		taio->used += rc;
+		if (taio->used == taio->size) {
+			td_async_io_callback(taio, taio->used, 0);
+			return;
+		}
+	}
+
+	td_async_io_update_timeout(taio);
+}
+
+static void td_async_io_writeable(event_id_t id, char mode, void *private)
+{
+	td_async_io_t *taio = private;
+	int rc;
+
+	while (1) {
+		rc = write(taio->fd, taio->buff + taio->used,
+			   taio->size - taio->used);
+
+		if (rc < 0) {
+			if (errno == EINTR)
+				continue;
+			if (errno == EWOULDBLOCK || errno == EAGAIN)
+				break;
+
+			td_async_io_callback(taio, 0, errno);
+			return;
+		}
+
+		taio->used += rc;
+		if (taio->used == taio->size) {
+			td_async_io_callback(taio, taio->used, 0);
+			return;
+		}
+	}
+
+	td_async_io_update_timeout(taio);
+}
+
+static void td_async_io_timeout(event_id_t id, char mode, void *private)
+{
+	td_async_io_t *taio = private;
+
+	td_async_io_kill(taio);
+	taio->callback(taio, 0, ETIME);
+}
+
+int td_async_io_is_running(td_async_io_t *taio)
+{
+	return taio->running;
+}
+
+void td_async_io_kill(td_async_io_t *taio)
+{
+	if (!taio->running)
+		return;
+
+	if (taio->timeout_id >= 0) {
+		tapdisk_server_unregister_event(taio->timeout_id);
+		taio->timeout_id = -1;
+	}
+
+	if (taio->io_id >= 0) {
+		tapdisk_server_unregister_event(taio->io_id);
+		taio->io_id = -1;
+	}
+
+	taio->running = 0;
+}
diff --git a/tools/blktap2/drivers/block-replication.h b/tools/blktap2/drivers/block-replication.h
index fdc216e..d39c530 100644
--- a/tools/blktap2/drivers/block-replication.h
+++ b/tools/blktap2/drivers/block-replication.h
@@ -156,6 +156,62 @@ struct hashtable *ramdisk_new_hashtable(void);
 int ramdisk_write_to_hashtable(struct hashtable *h, uint64_t sector,
 			       int nb_sectors, size_t sector_size, char* buf,
 			       const char *log_prefix);
+int ramdisk_read_from_hashtable(struct hashtable *h, uint64_t sector,
+				int nb_sectors, int sector_size,
+				char *buf);
 void ramdisk_destroy_hashtable(struct hashtable *h);
 
+/* async I/O, don't support read/write at the same time */
+typedef struct td_async_io td_async_io_t;
+enum {
+	td_async_read,
+	td_async_write,
+};
+
+/*
+ * realsize >= 1 means all data was read/written
+ * realsize == 0 means failure happened when reading/writing, and
+ * errnoval is valid
+ * realsize == -1 means some other internal failure happended, and
+ * errnoval is also valid
+ * In all cases async_io is killed before calling this callback
+ *
+ * If we don't read/write any more data in timeout_s seconds, realsize is
+ * 0, and errnoval is ETIME
+ *
+ * If timeout_s is 0, timeout will be disabled.
+ *
+ * NOTE: realsize is less than taio->size, if we read EOF.
+ */
+typedef void taio_callback(td_async_io_t *taio, int realsize,
+			   int errnoval);
+
+struct td_async_io {
+	/* caller must fill these in, and they must all remain valid */
+	int fd;
+	int timeout_s;
+	int mode;
+	/*
+	 * read: store the data to buff
+	 * write: point to the data to be written
+	 */
+	void *buff;
+	int size;
+	taio_callback *callback;
+
+	/* private */
+	event_id_t timeout_id, io_id;
+	int used;
+	int running;
+};
+
+/* Don't call it when td_async_io is running */
+void td_async_io_init(td_async_io_t *taio);
+/* return -1 if we find some error. Otherwise, return 0 */
+int td_async_io_start(td_async_io_t *taio);
+/* return 1 if td_async_io is running, otherwise return 0 */
+int td_async_io_is_running(td_async_io_t *taio);
+/* The callback will not be called */
+void td_async_io_kill(td_async_io_t *taio);
+
 #endif
diff --git a/tools/blktap2/drivers/tapdisk-disktype.c b/tools/blktap2/drivers/tapdisk-disktype.c
index 8d1383b..aa2afab 100644
--- a/tools/blktap2/drivers/tapdisk-disktype.c
+++ b/tools/blktap2/drivers/tapdisk-disktype.c
@@ -94,6 +94,12 @@ static const disk_info_t remus_disk = {
        0,
 };
 
+static const disk_info_t colo_disk = {
+	"colo",
+	"colo disk replicator (COLO)",
+	0,
+};
+
 const disk_info_t *tapdisk_disk_types[] = {
 	[DISK_TYPE_AIO]	= &aio_disk,
 	[DISK_TYPE_SYNC]	= &sync_disk,
@@ -105,6 +111,7 @@ const disk_info_t *tapdisk_disk_types[] = {
 	[DISK_TYPE_BLOCK_CACHE] = &block_cache_disk,
 	[DISK_TYPE_LOG]	= &log_disk,
 	[DISK_TYPE_REMUS]	= &remus_disk,
+	[DISK_TYPE_COLO]	= &colo_disk,
 	[DISK_TYPE_MAX]		= NULL,
 };
 
@@ -119,6 +126,7 @@ extern struct tap_disk tapdisk_block_cache;
 extern struct tap_disk tapdisk_vhd_index;
 extern struct tap_disk tapdisk_log;
 extern struct tap_disk tapdisk_remus;
+extern struct tap_disk tapdisk_colo;
 
 const struct tap_disk *tapdisk_disk_drivers[] = {
 	[DISK_TYPE_AIO]         = &tapdisk_aio,
@@ -132,6 +140,7 @@ const struct tap_disk *tapdisk_disk_drivers[] = {
 	[DISK_TYPE_BLOCK_CACHE] = &tapdisk_block_cache,
 	[DISK_TYPE_LOG]         = &tapdisk_log,
 	[DISK_TYPE_REMUS]       = &tapdisk_remus,
+	[DISK_TYPE_COLO]        = &tapdisk_colo,
 	[DISK_TYPE_MAX]         = NULL,
 };
 
diff --git a/tools/blktap2/drivers/tapdisk-disktype.h b/tools/blktap2/drivers/tapdisk-disktype.h
index c574990..ee8cb02 100644
--- a/tools/blktap2/drivers/tapdisk-disktype.h
+++ b/tools/blktap2/drivers/tapdisk-disktype.h
@@ -39,7 +39,8 @@
 #define DISK_TYPE_BLOCK_CACHE 7
 #define DISK_TYPE_LOG         8
 #define DISK_TYPE_REMUS       9
-#define DISK_TYPE_MAX         10
+#define DISK_TYPE_COLO        10
+#define DISK_TYPE_MAX         11
 
 #define DISK_TYPE_NAME_MAX    32
 
-- 
1.9.3