[PATCH v12 5/6] migrate: Migrate TCP flows

[Stefano Brivio <sbrivio@redhat.com>]

This implements flow preparation on the source, transfer of data with
a format roughly inspired by struct tcp_tap_conn, and flow insertion
on the target, with all the appropriate window options, window
scaling, MSS, etc.

The target side is rather convoluted because we first need to create
sockets and switch them to repair mode, before we can apply options
that are *not* stored in the flow table. However, we don't want to
request repair mode for sockets one by one. So we need to do this in
several steps.

[dwg: Assorted cleanups]
Signed-off-by: David Gibson <david@gibson.dropbear.id.au>
Signed-off-by: Stefano Brivio <sbrivio@redhat.com>
---
 flow.c     | 202 +++++++++++++++++++
 flow.h     |   6 +
 migrate.c  |  10 +
 repair.c   |   1 -
 tcp.c      | 583 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 tcp_conn.h |  87 ++++++++
 6 files changed, 888 insertions(+), 1 deletion(-)

diff --git a/flow.c b/flow.c
index a6fe6d1..aca257f 100644
--- a/flow.c
+++ b/flow.c
@@ -19,6 +19,7 @@
 #include "inany.h"
 #include "flow.h"
 #include "flow_table.h"
+#include "repair.h"
 
 const char *flow_state_str[] = {
 	[FLOW_STATE_FREE]	= "FREE",
@@ -52,6 +53,26 @@ const uint8_t flow_proto[] = {
 static_assert(ARRAY_SIZE(flow_proto) == FLOW_NUM_TYPES,
 	      "flow_proto[] doesn't match enum flow_type");
 
+#define foreach_flow(i, flow, bound)					\
+	for ((i) = 0, (flow) = &flowtab[(i)];				\
+	     (i) < (bound);						\
+	     (i)++, (flow) = &flowtab[(i)])				\
+		if ((flow)->f.state == FLOW_STATE_FREE)			\
+			(i) += (flow)->free.n - 1;			\
+		else
+
+#define foreach_active_flow(i, flow, bound)				\
+	foreach_flow((i), (flow), (bound))				\
+		if ((flow)->f.state != FLOW_STATE_ACTIVE)		\
+			continue;					\
+		else
+
+#define foreach_tcp_flow(i, flow, bound)				\
+	foreach_active_flow((i), (flow), (bound))			\
+		if ((flow)->f.type != FLOW_TCP)				\
+			continue;					\
+		else
+
 /* Global Flow Table */
 
 /**
@@ -874,6 +895,187 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now)
 	*last_next = FLOW_MAX;
 }
 
+/**
+ * flow_migrate_source_pre_do() - Prepare/"unprepare" source flows for migration
+ * @c:		Execution context
+ * @stage:	Migration stage information (unused)
+ * @fd:		Migration file descriptor (unused)
+ * @rollback:	If true, undo preparation
+ *
+ * Return: 0 on success, positive error code on failure
+ */
+static int flow_migrate_source_pre_do(struct ctx *c,
+				      const struct migrate_stage *stage, int fd,
+				      bool rollback)
+{
+	unsigned i, max_i;
+	union flow *flow;
+	int rc = 0;
+
+	(void)stage;
+	(void)fd;
+
+	if (rollback) {
+		i = FLOW_MAX;
+		goto rollback;
+	}
+
+	foreach_tcp_flow(i, flow, FLOW_MAX) {
+		rc = tcp_flow_repair_on(c, &flow->tcp);
+		if (rc) {
+			debug("Can't repair TCP flows: %s, roll back",
+			      strerror_(-rc));
+			goto rollback;
+		}
+	}
+
+	if ((rc = repair_flush(c))) {
+		debug("Can't repair TCP flows: %s, roll back",
+		      strerror_(-rc));
+		goto rollback;
+	}
+
+	return 0;
+
+rollback:
+	max_i = i;
+
+	foreach_tcp_flow(i, flow, max_i)
+		tcp_flow_repair_off(c, &flow->tcp);
+
+	repair_flush(c);
+
+	return -rc;
+}
+
+/**
+ * flow_migrate_source_pre() - Prepare source flows for migration
+ * @c:		Execution context
+ * @stage:	Migration stage information
+ * @fd:		Migration file descriptor
+ * @rollback:	If true, undo preparation
+ *
+ * Return: 0 on success, positive error code on failure
+ */
+int flow_migrate_source_pre(struct ctx *c, const struct migrate_stage *stage,
+			    int fd)
+{
+	return flow_migrate_source_pre_do(c, stage, fd, false);
+}
+
+/**
+ * flow_migrate_source() - Dump additional information and send data
+ * @c:		Execution context
+ * @stage:	Migration stage information (unused)
+ * @fd:		Migration file descriptor
+ *
+ * Return: 0 on success, positive error code on failure
+ */
+int flow_migrate_source(struct ctx *c, const struct migrate_stage *stage,
+			int fd)
+{
+	uint32_t count = 0;
+	union flow *flow;
+	unsigned i;
+	int rc;
+
+	(void)stage;
+
+	foreach_tcp_flow(i, flow, FLOW_MAX)
+		count++;
+
+	count = htonl(count);
+	rc = write_all_buf(fd, &count, sizeof(count));
+	if (rc) {
+		rc = errno;
+		err_perror("Can't send flow count (%u), abort", ntohl(count));
+		return rc;
+	}
+	debug("Sending %u flows", ntohl(count));
+
+	/* Send information that can be stored in the flow table, first */
+	foreach_tcp_flow(i, flow, FLOW_MAX) {
+		rc = tcp_flow_migrate_source(fd, &flow->tcp);
+		if (rc) {
+			err("Can't send data, flow %u: %s, roll back",
+			    i, strerror_(-rc));
+			goto rollback;
+		}
+	}
+
+	/* TODO: other protocols */
+
+	/* And then "extended" data: the target needs to set repair mode on
+	 * sockets before it can set this stuff, but it needs sockets (and
+	 * flows) for that.
+	 */
+	foreach_tcp_flow(i, flow, FLOW_MAX) {
+		rc = tcp_flow_migrate_source_ext(fd, &flow->tcp);
+		if (rc) {
+			err("Can't send extended data, flow %u: %s, roll back",
+			    i, strerror_(-rc));
+			goto rollback;
+		}
+	}
+
+	/* TODO: other protocols */
+
+	return 0;
+
+rollback:
+	flow_migrate_source_pre_do(c, stage, fd, true);
+	return -rc;
+}
+
+/**
+ * flow_migrate_target() - Receive flows and insert in flow table
+ * @c:		Execution context
+ * @stage:	Migration stage information (unused)
+ * @fd:		Migration file descriptor
+ *
+ * Return: 0 on success, positive error code on failure
+ */
+int flow_migrate_target(struct ctx *c, const struct migrate_stage *stage,
+			int fd)
+{
+	uint32_t count;
+	unsigned i;
+	int rc;
+
+	(void)stage;
+
+	if (read_all_buf(fd, &count, sizeof(count)))
+		return errno;
+
+	count = ntohl(count);
+	debug("Receiving %u flows", count);
+
+	/* TODO: flow header with type, instead? */
+	for (i = 0; i < count; i++) {
+		rc = tcp_flow_migrate_target(c, fd);
+		if (rc) {
+			debug("Bad target data for flow %u: %s, abort",
+			      i, strerror_(-rc));
+			return -rc;
+		}
+	}
+
+	repair_flush(c);
+
+	for (i = 0; i < count; i++) {
+		rc = tcp_flow_migrate_target_ext(c, flowtab + i, fd);
+		if (rc) {
+			debug("Bad target extended data for flow %u: %s, abort",
+			      i, strerror_(-rc));
+			return -rc;
+		}
+	}
+
+	repair_flush(c);
+
+	return 0;
+}
+
 /**
  * flow_init() - Initialise flow related data structures
  */
diff --git a/flow.h b/flow.h
index 24ba3ef..a485c35 100644
--- a/flow.h
+++ b/flow.h
@@ -249,6 +249,12 @@ union flow;
 
 void flow_init(void);
 void flow_defer_handler(const struct ctx *c, const struct timespec *now);
+int flow_migrate_source_pre(struct ctx *c, const struct migrate_stage *stage,
+			    int fd);
+int flow_migrate_source(struct ctx *c, const struct migrate_stage *stage,
+			int fd);
+int flow_migrate_target(struct ctx *c, const struct migrate_stage *stage,
+			int fd);
 
 void flow_log_(const struct flow_common *f, int pri, const char *fmt, ...)
 	__attribute__((format(printf, 3, 4)));
diff --git a/migrate.c b/migrate.c
index 1c59016..0fca77b 100644
--- a/migrate.c
+++ b/migrate.c
@@ -103,6 +103,16 @@ static const struct migrate_stage stages_v1[] = {
 		.source = seen_addrs_source_v1,
 		.target = seen_addrs_target_v1,
 	},
+	{
+		.name = "prepare flows",
+		.source = flow_migrate_source_pre,
+		.target = NULL,
+	},
+	{
+		.name = "transfer flows",
+		.source = flow_migrate_source,
+		.target = flow_migrate_target,
+	},
 	{ 0 },
 };
 
diff --git a/repair.c b/repair.c
index 784b994..da85edb 100644
--- a/repair.c
+++ b/repair.c
@@ -190,7 +190,6 @@ int repair_flush(struct ctx *c)
  *
  * Return: 0 on success, negative error code on failure
  */
-/* cppcheck-suppress unusedFunction */
 int repair_set(struct ctx *c, int s, int cmd)
 {
 	int rc;
diff --git a/tcp.c b/tcp.c
index af6bd95..7367244 100644
--- a/tcp.c
+++ b/tcp.c
@@ -299,6 +299,7 @@
 #include "log.h"
 #include "inany.h"
 #include "flow.h"
+#include "repair.h"
 #include "linux_dep.h"
 
 #include "flow_table.h"
@@ -326,6 +327,14 @@
 	 ((conn)->events & (SOCK_FIN_RCVD | TAP_FIN_RCVD)))
 #define CONN_HAS(conn, set)	(((conn)->events & (set)) == (set))
 
+/* Buffers to migrate pending data from send and receive queues. No, they don't
+ * use memory if we don't use them.
+ */
+#define TCP_MIGRATE_SND_QUEUE_MAX	(16 << 20)
+#define TCP_MIGRATE_RCV_QUEUE_MAX	(16 << 20)
+uint8_t tcp_migrate_snd_queue[TCP_MIGRATE_SND_QUEUE_MAX];
+uint8_t tcp_migrate_rcv_queue[TCP_MIGRATE_RCV_QUEUE_MAX];
+
 static const char *tcp_event_str[] __attribute((__unused__)) = {
 	"SOCK_ACCEPTED", "TAP_SYN_RCVD", "ESTABLISHED", "TAP_SYN_ACK_SENT",
 
@@ -2645,3 +2654,577 @@ void tcp_timer(struct ctx *c, const struct timespec *now)
 	if (c->mode == MODE_PASTA)
 		tcp_splice_refill(c);
 }
+
+/**
+ * tcp_flow_repair_on() - Enable repair mode for a single TCP flow
+ * @c:		Execution context
+ * @conn:	Pointer to the TCP connection structure
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_repair_on(struct ctx *c, const struct tcp_tap_conn *conn)
+{
+	int rc = 0;
+
+	if ((rc = repair_set(c, conn->sock, TCP_REPAIR_ON)))
+		err("Failed to set TCP_REPAIR");
+
+	return rc;
+}
+
+/**
+ * tcp_flow_repair_off() - Clear repair mode for a single TCP flow
+ * @c:		Execution context
+ * @conn:	Pointer to the TCP connection structure
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_repair_off(struct ctx *c, const struct tcp_tap_conn *conn)
+{
+	int rc = 0;
+
+	if ((rc = repair_set(c, conn->sock, TCP_REPAIR_OFF)))
+		err("Failed to clear TCP_REPAIR");
+
+	return rc;
+}
+
+/**
+ * tcp_flow_repair_queues() - Dump or set sequences, read or write socket queues
+ * @s:		Socket
+ * @snd_seq:	Send sequence, set on return if @set == false, network order
+ * @snd_buf:	Send queue buffer, read or written depending on @set
+ * @snd_len:	Length of send queue buffer, network order
+ * @rcv_seq:	Receive sequence, set on return if @set == false, network order
+ * @rcv_buf:	Receive queue buffer, read or written depending on @set
+ * @rcv_len:	Length of receive queue buffer, network order
+ * @set:	Set if true, dump if false
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+static int tcp_flow_repair_queues(int s,
+				  uint32_t *snd_seq, uint8_t *snd_buf,
+				  uint32_t *snd_len,
+				  uint32_t *rcv_seq, uint8_t *rcv_buf,
+				  uint32_t *rcv_len, bool set)
+{
+	socklen_t vlen = sizeof(uint32_t);
+	ssize_t rc;
+	int v;
+
+	v = TCP_SEND_QUEUE;
+	if (setsockopt(s, SOL_TCP, TCP_REPAIR_QUEUE, &v, sizeof(v))) {
+		rc = -errno;
+		err_perror("Selecting TCP_SEND_QUEUE on socket %i", s);
+		return rc;
+	}
+
+	if (set) {
+		uint8_t *p;
+
+		*snd_seq = ntohl(*snd_seq);
+		if (setsockopt(s, SOL_TCP, TCP_QUEUE_SEQ, snd_seq, vlen)) {
+			rc = -errno;
+			err_perror("Setting send sequence for socket %i", s);
+			return rc;
+		}
+		debug("Set send sequence for socket %i to %u", s, *snd_seq);
+
+		debug("Writing socket %i send queue: %u bytes", s, *snd_len);
+		p = snd_buf;
+		while (*snd_len > 0) {
+			rc = send(s, p, *snd_len, 0);
+
+			if (rc < 0) {
+				rc = -errno;
+				err_perror("Can't write socket %i send queue",
+					   s);
+				return rc;
+			}
+
+			snd_len -= rc;
+			p += rc;
+		}
+	} else {
+		if (getsockopt(s, SOL_TCP, TCP_QUEUE_SEQ, snd_seq, &vlen)) {
+			rc = -errno;
+			err_perror("Dumping send sequence for socket %i", s);
+			return rc;
+		}
+		debug("Dumped send sequence for socket %i: %u", s, *snd_seq);
+		*snd_seq = htonl(*snd_seq);
+
+		rc = recv(s, snd_buf, TCP_MIGRATE_SND_QUEUE_MAX, MSG_PEEK);
+		if (rc < 0 && errno != EAGAIN) { /* EAGAIN means empty */
+			rc = -errno;
+			err_perror("Can't read send queue for socket %i", s);
+			return rc;
+		}
+
+		rc = MAX(0, rc);
+		*snd_len = htonl(rc);
+		debug("Read socket %i send queue: %zi bytes", s, rc);
+	}
+
+	v = TCP_RECV_QUEUE;
+	if (setsockopt(s, SOL_TCP, TCP_REPAIR_QUEUE, &v, sizeof(v))) {
+		rc = -errno;
+		err_perror("Selecting TCP_RECV_QUEUE for socket %i", s);
+		return rc;
+	}
+
+	if (set) {
+		uint8_t *p;
+
+		*rcv_seq = ntohl(*rcv_seq);
+		if (setsockopt(s, SOL_TCP, TCP_QUEUE_SEQ, rcv_seq, vlen)) {
+			rc = -errno;
+			err_perror("Setting receive sequence %u for socket %i",
+				   *rcv_seq, s);
+			return rc;
+		}
+		debug("Set receive sequence for socket %i to %u", s, *rcv_seq);
+
+		debug("Writing socket %i receive queue: %u bytes", s, *rcv_len);
+		p = rcv_buf;
+		while (*rcv_len > 0) {
+			rc = send(s, p, *rcv_len, 0);
+
+			if (rc < 0) {
+				rc = -errno;
+				err_perror("Can't send socket %i receive queue",
+					   s);
+				return rc;
+			}
+
+			rcv_len -= rc;
+			p += rc;
+		}
+	} else {
+		if (getsockopt(s, SOL_TCP, TCP_QUEUE_SEQ, rcv_seq, &vlen)) {
+			rc = -errno;
+			err_perror("Dumping receive sequence for socket %i", s);
+			return rc;
+		}
+		debug("Dumped receive sequence for socket %i: %u", s, *rcv_seq);
+		*rcv_seq = htonl(*rcv_seq);
+
+		rc = recv(s, rcv_buf, TCP_MIGRATE_RCV_QUEUE_MAX, MSG_PEEK);
+		if (rc < 0 && errno != EAGAIN) { /* EAGAIN means empty */
+			rc = -errno;
+			err_perror("Can't read receive queue for socket %i", s);
+			return rc;
+		}
+
+		rc = MAX(0, rc);
+		*rcv_len = htonl(rc);
+		debug("Read socket %i receive queue: %zi bytes", s, rc);
+	}
+
+	return 0;
+}
+
+/**
+ * tcp_flow_repair_opt() - Dump or set repair "options" (MSS and window scale)
+ * @s:			Socket
+ * @snd_wscale:		Window scaling factor, send, network order
+ * @rcv_wscale:		Window scaling factor, receive, network order
+ * @mss:		Maximum Segment Size, socket side, network order
+ * @set:		Set if true, dump if false
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_repair_opt(int s, uint8_t *snd_wscale, uint8_t *rcv_wscale,
+			uint32_t *mss, bool set)
+{
+	struct tcp_info_linux tinfo;
+	struct tcp_repair_opt opts[2];
+	socklen_t sl;
+	int rc;
+
+	opts[0].opt_code = TCPOPT_WINDOW;
+	opts[1].opt_code = TCPOPT_MAXSEG;
+
+	if (set) {
+		debug("Setting repair options for socket %i:", s);
+		opts[0].opt_val = *snd_wscale + (*rcv_wscale << 16);
+		opts[1].opt_val = ntohl(*mss);
+		debug("  window scale send %u, receive %u, MSS: %u",
+		      *snd_wscale, *rcv_wscale, ntohl(*mss));
+
+		sl = sizeof(opts);
+		if (setsockopt(s, SOL_TCP, TCP_REPAIR_OPTIONS, opts, sl)) {
+			rc = -errno;
+			err_perror("Setting repair options for socket %i", s);
+			return rc;
+		}
+	} else {
+		sl = sizeof(tinfo);
+		if (getsockopt(s, SOL_TCP, TCP_INFO, &tinfo, &sl)) {
+			rc = -errno;
+			err_perror("Querying TCP_INFO for socket %i", s);
+			return rc;
+		}
+
+		*snd_wscale = tinfo.tcpi_snd_wscale;
+		*rcv_wscale = tinfo.tcpi_rcv_wscale;
+
+		/* TCP_INFO MSS value not always in sync: query explicitly */
+		sl = sizeof(*mss);
+		if (getsockopt(s, SOL_TCP, TCP_MAXSEG, mss, &sl)) {
+			rc = -errno;
+			err_perror("Getting MSS for socket %i", s);
+			return rc;
+		}
+		*mss = htonl(*mss);
+
+		debug("Got repair options for socket %i:", s);
+		debug("  window scale send %u, receive %u, MSS: %u",
+		      *snd_wscale, *rcv_wscale, ntohl(*mss));
+	}
+
+	return 0;
+}
+
+/**
+ * tcp_flow_repair_wnd() - Dump or set window parameters
+ * @snd_wl1:		Next sequence used in window probe (next sequence - 1)
+ * @snd_wnd:		Socket-side sending window, network order
+ * @max_window:		Window clamp, network order
+ * @rcv_wnd:		Socket-side receive window, network order
+ * @rcv_wup:		rcv_nxt on last window update sent, network order
+ * @set:		Set if true, dump if false
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_repair_wnd(int s, uint32_t *snd_wl1, uint32_t *snd_wnd,
+			uint32_t *max_window, uint32_t *rcv_wnd,
+			uint32_t *rcv_wup, bool set)
+{
+	struct tcp_repair_window wnd;
+	socklen_t sl = sizeof(wnd);
+	int rc;
+
+	if (set) {
+		wnd.snd_wl1	= ntohl(*snd_wl1);
+		wnd.snd_wnd	= ntohl(*snd_wnd);
+		wnd.max_window	= ntohl(*max_window);
+		wnd.rcv_wnd	= ntohl(*rcv_wnd);
+		wnd.rcv_wup	= ntohl(*rcv_wup);
+
+		if (setsockopt(s, IPPROTO_TCP, TCP_REPAIR_WINDOW, &wnd, sl)) {
+			rc = -errno;
+			err_perror("Setting window repair data, socket %i", s);
+			return rc;
+		}
+	} else {
+		if (getsockopt(s, IPPROTO_TCP, TCP_REPAIR_WINDOW, &wnd, &sl)) {
+			rc = -errno;
+			err_perror("Getting window repair data, socket %i", s);
+			return rc;
+		}
+
+		*snd_wl1	= htonl(wnd.snd_wl1);
+		*snd_wnd	= htonl(wnd.snd_wnd);
+		*max_window	= htonl(wnd.max_window);
+		*rcv_wnd	= htonl(wnd.rcv_wnd);
+		*rcv_wup	= htonl(wnd.rcv_wup);
+	}
+
+	return 0;
+}
+
+/**
+ * tcp_flow_migrate_source() - Send data (flow table part) for a single flow
+ * @c:		Execution context
+ * @fd:		Descriptor for state migration
+ * @conn:	Pointer to the TCP connection structure
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_migrate_source(int fd, const struct tcp_tap_conn *conn)
+{
+	int rc;
+
+	struct tcp_tap_transfer t = {
+		.retrans		= conn->retrans,
+		.ws_from_tap		= conn->ws_from_tap,
+		.ws_to_tap		= conn->ws_to_tap,
+		.events			= conn->events,
+
+		.tap_mss		= htonl(MSS_GET(conn)),
+
+		.sndbuf			= htonl(conn->sndbuf),
+
+		.flags			= conn->flags,
+		.seq_dup_ack_approx	= conn->seq_dup_ack_approx,
+
+		.wnd_from_tap		= htons(conn->wnd_from_tap),
+		.wnd_to_tap		= htons(conn->wnd_to_tap),
+
+		.seq_to_tap		= htonl(conn->seq_to_tap),
+		.seq_ack_from_tap	= htonl(conn->seq_ack_from_tap),
+		.seq_from_tap		= htonl(conn->seq_from_tap),
+		.seq_ack_to_tap		= htonl(conn->seq_ack_to_tap),
+		.seq_init_from_tap	= htonl(conn->seq_init_from_tap),
+	};
+
+	memcpy(&t.pif, conn->f.pif, sizeof(t.pif));
+	memcpy(&t.side, conn->f.side, sizeof(t.side));
+
+	if (write_all_buf(fd, &t, sizeof(t))) {
+		rc = -errno;
+		err_perror("Failed to write migration data for socket %i",
+			   conn->sock);
+		return rc;
+	}
+
+	return 0;
+}
+
+/**
+ * tcp_flow_migrate_source_ext() - Send extended data for a single flow
+ * @fd:		Descriptor for state migration
+ * @conn:	Pointer to the TCP connection structure
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_migrate_source_ext(int fd, const struct tcp_tap_conn *conn)
+{
+	struct tcp_tap_transfer_ext t;
+	int s = conn->sock;
+	int rc;
+
+	rc = tcp_flow_repair_queues(s,
+				    &t.sock_seq_snd, tcp_migrate_snd_queue,
+				    &t.sndlen,
+				    &t.sock_seq_rcv, tcp_migrate_rcv_queue,
+				    &t.rcvlen, false);
+	if (rc) {
+		err("Failed to repair queues on source for socket %i: %s",
+		    s, strerror_(-rc));
+		return rc;
+	}
+
+	tcp_flow_repair_opt(s, &t.snd_wscale, &t.rcv_wscale, &t.sock_mss,
+			    false);
+
+	tcp_flow_repair_wnd(s, &t.sock_snd_wl1, &t.sock_snd_wnd,
+			    &t.sock_max_window, &t.sock_rcv_wnd,
+			    &t.sock_rcv_wup, false);
+
+	if (write_all_buf(fd, &t, sizeof(t))) {
+		rc = -errno;
+		err_perror("Failed to write extended data for socket %i", s);
+		return rc;
+	}
+
+	if (write_all_buf(fd, tcp_migrate_snd_queue, ntohl(t.sndlen))) {
+		rc = -errno;
+		err_perror("Failed to write send queue data for socket %i", s);
+		return rc;
+	}
+
+	if (write_all_buf(fd, tcp_migrate_rcv_queue, ntohl(t.rcvlen))) {
+		rc = -errno;
+		err_perror("Failed to write receive queue data for socket %i",
+			   s);
+		return rc;
+	}
+
+	return 0;
+}
+
+/**
+ * tcp_flow_repair_socket() - Open and bind socket, request repair mode
+ * @c:		Execution context
+ * @conn:	Pointer to the TCP connection structure
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_repair_socket(struct ctx *c, struct tcp_tap_conn *conn)
+{
+	sa_family_t af = CONN_V4(conn) ? AF_INET : AF_INET6;
+	const struct flowside *sockside = HOSTFLOW(conn);
+	union sockaddr_inany a;
+	socklen_t sl;
+	int s, rc;
+
+	pif_sockaddr(c, &a, &sl, PIF_HOST, &sockside->oaddr, sockside->oport);
+
+	if ((s = conn->sock = socket(af, SOCK_STREAM, IPPROTO_TCP)) < 0) {
+		rc = -errno;
+		err_perror("Failed to create socket for migrated flow");
+		return rc;
+	}
+
+	if (setsockopt(s, SOL_TCP, TCP_NODELAY, &((int){ 1 }), sizeof(int)))
+		debug_perror("Failed to set TCP_NODELAY for socket %i", s);
+
+	if (bind(s, &a.sa, sizeof(a)) < 0) {
+		rc = -errno;
+		err_perror("Failed to bind socket %i for migrated flow", s);
+		close(s);
+		conn->sock = -1;
+		return rc;
+	}
+
+	rc = tcp_flow_repair_on(c, conn);
+	if (rc) {
+		close(s);
+		conn->sock = -1;
+		return rc;
+	}
+
+	return 0;
+}
+
+/**
+ * tcp_flow_repair_connect() - Connect socket in repair mode, then turn it off
+ * @c:		Execution context
+ * @conn:	Pointer to the TCP connection structure
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+static int tcp_flow_repair_connect(const struct ctx *c,
+				   struct tcp_tap_conn *conn)
+{
+	const struct flowside *tgt = HOSTFLOW(conn);
+	int rc;
+
+	if ((rc = flowside_connect(c, conn->sock, PIF_HOST, tgt))) {
+		err("Failed to connect migrated socket %i: %s", conn->sock, 
+		    strerror_(-rc));
+		return rc;
+	}
+
+	conn->in_epoll = 0;
+	conn->timer = -1;
+	if ((rc = tcp_epoll_ctl(c, conn))) {
+		debug("Failed to subscribe to epoll for migrated socket %i: %s",
+		      conn->sock, strerror_(-rc));
+	}
+
+	return 0;
+}
+
+/**
+ * tcp_flow_migrate_target() - Receive data (flow table part) for flow, insert
+ * @c:		Execution context
+ * @fd:		Descriptor for state migration
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_migrate_target(struct ctx *c, int fd)
+{
+	struct tcp_tap_transfer t;
+	struct tcp_tap_conn *conn;
+	union flow *flow;
+	int rc;
+
+	if (!(flow = flow_alloc())) {
+		err("Flow table full on migration target");
+		return -ENOMEM;
+	}
+
+	if (read_all_buf(fd, &t, sizeof(t))) {
+		err_perror("Failed to receive migration data");
+		return -errno;
+	}
+
+	flow->f.state = FLOW_STATE_TGT;
+	memcpy(&flow->f.pif, &t.pif, sizeof(flow->f.pif));
+	memcpy(&flow->f.side, &t.side, sizeof(flow->f.side));
+	conn = FLOW_SET_TYPE(flow, FLOW_TCP, tcp);
+
+	conn->retrans			= t.retrans;
+	conn->ws_from_tap		= t.ws_from_tap;
+	conn->ws_to_tap			= t.ws_to_tap;
+	conn->events			= t.events;
+
+	conn->sndbuf			= htonl(t.sndbuf);
+
+	conn->flags			= t.flags;
+	conn->seq_dup_ack_approx	= t.seq_dup_ack_approx;
+
+	MSS_SET(conn,			  ntohl(t.tap_mss));
+
+	conn->wnd_from_tap		= ntohs(t.wnd_from_tap);
+	conn->wnd_to_tap		= ntohs(t.wnd_to_tap);
+
+	conn->seq_to_tap		= ntohl(t.seq_to_tap);
+	conn->seq_ack_from_tap		= ntohl(t.seq_ack_from_tap);
+	conn->seq_from_tap		= ntohl(t.seq_from_tap);
+	conn->seq_ack_to_tap		= ntohl(t.seq_ack_to_tap);
+	conn->seq_init_from_tap		= ntohl(t.seq_init_from_tap);
+
+	if ((rc = tcp_flow_repair_socket(c, conn)))
+		return rc;
+
+	flow_hash_insert(c, TAP_SIDX(conn));
+	FLOW_ACTIVATE(conn);
+
+	return 0;
+}
+
+/**
+ * tcp_flow_migrate_target_ext() - Receive extended data for flow, set, connect
+ * @c:		Execution context
+ * @flow:	Existing flow for this connection data
+ * @fd:		Descriptor for state migration
+ *
+ * Return: 0 on success, negative code on failure, but 0 on connection reset
+ */
+int tcp_flow_migrate_target_ext(struct ctx *c, union flow *flow, int fd)
+{
+	struct tcp_tap_conn *conn = &flow->tcp;
+	struct tcp_tap_transfer_ext t;
+	int s = conn->sock, rc;
+	uint32_t peek_offset;
+
+	if (read_all_buf(fd, &t, sizeof(t))) {
+		rc = -errno;
+		err_perror("Failed to read extended data for socket %i", s);
+		return rc;
+	}
+
+	if (read_all_buf(fd, tcp_migrate_snd_queue, ntohl(t.sndlen))) {
+		rc = -errno;
+		err_perror("Failed to read send queue data for socket %i", s);
+		return rc;
+	}
+
+	if (read_all_buf(fd, tcp_migrate_rcv_queue, ntohl(t.rcvlen))) {
+		rc = -errno;
+		err_perror("Failed to read receive queue data for socket %i",
+			   s);
+		return rc;
+	}
+
+	if ((rc = tcp_flow_repair_queues(s,
+					 &t.sock_seq_snd, tcp_migrate_snd_queue,
+					 &t.sndlen,
+					 &t.sock_seq_rcv, tcp_migrate_rcv_queue,
+				         &t.rcvlen, true)))
+		return rc;
+
+	if ((rc = tcp_flow_repair_connect(c, conn)))
+		return rc;
+
+	if ((rc = tcp_flow_repair_wnd(s, &t.sock_snd_wl1, &t.sock_snd_wnd,
+				      &t.sock_max_window, &t.sock_rcv_wnd,
+				      &t.sock_rcv_wup, true)))
+		return rc;
+
+	if ((rc = tcp_flow_repair_opt(s, &t.snd_wscale, &t.rcv_wscale,
+				      &t.sock_mss, true)))
+		return rc;
+
+	tcp_flow_repair_off(c, conn);
+
+	peek_offset = conn->seq_to_tap - conn->seq_ack_from_tap;
+	if (tcp_set_peek_offset(conn->sock, peek_offset))
+		tcp_rst(c, conn);
+
+	return 0;
+}
diff --git a/tcp_conn.h b/tcp_conn.h
index d342680..d810e6a 100644
--- a/tcp_conn.h
+++ b/tcp_conn.h
@@ -96,6 +96,87 @@ struct tcp_tap_conn {
 	uint32_t	seq_init_from_tap;
 };
 
+/**
+ * struct tcp_tap_transfer - Migrated TCP data, flow table part, network order
+ * @pif:		Interfaces for each side of the flow
+ * @side:		Addresses and ports for each side of the flow
+ * @retrans:		Number of retransmissions occurred due to ACK_TIMEOUT
+ * @ws_from_tap:	Window scaling factor advertised from tap/guest
+ * @ws_to_tap:		Window scaling factor advertised to tap/guest
+ * @events:		Connection events, implying connection states
+ * @tap_mss:		MSS advertised by tap/guest, rounded to 2 ^ TCP_MSS_BITS
+ * @sndbuf:		Sending buffer in kernel, rounded to 2 ^ SNDBUF_BITS
+ * @flags:		Connection flags representing internal attributes
+ * @seq_dup_ack_approx:	Last duplicate ACK number sent to tap
+ * @wnd_from_tap:	Last window size from tap, unscaled (as received)
+ * @wnd_to_tap:		Sending window advertised to tap, unscaled (as sent)
+ * @seq_to_tap:		Next sequence for packets to tap
+ * @seq_ack_from_tap:	Last ACK number received from tap
+ * @seq_from_tap:	Next sequence for packets from tap (not actually sent)
+ * @seq_ack_to_tap:	Last ACK number sent to tap
+ * @seq_init_from_tap:	Initial sequence number from tap
+*/
+struct tcp_tap_transfer {
+	uint8_t		pif[SIDES];
+	struct flowside	side[SIDES];
+
+	uint8_t		retrans;
+	uint8_t		ws_from_tap;
+	uint8_t		ws_to_tap;
+	uint8_t		events;
+
+	uint32_t	tap_mss;
+
+	uint32_t	sndbuf;
+
+	uint8_t		flags;
+	uint8_t		seq_dup_ack_approx;
+
+	uint16_t	wnd_from_tap;
+	uint16_t	wnd_to_tap;
+
+	uint32_t	seq_to_tap;
+	uint32_t	seq_ack_from_tap;
+	uint32_t	seq_from_tap;
+	uint32_t	seq_ack_to_tap;
+	uint32_t	seq_init_from_tap;
+} __attribute__((packed, aligned(__alignof__(uint32_t))));
+
+/**
+ * struct tcp_tap_transfer_ext - Migrated TCP data, outside flow, network order
+ * @sock_seq_snd:	Socket-side send sequence
+ * @sock_seq_rcv:	Socket-side receive sequence
+ * @sndlen:		Length of pending send queue
+ * @rcvlen:		Length of pending receive queue
+ * @sock_mss:		Socket-side MSS
+ * @sock_snd_wl1:	Next sequence used in window probe (next sequence - 1)
+ * @sock_snd_wnd:	Socket-side sending window
+ * @sock_max_window:	Window clamp
+ * @sock_rcv_wnd:	Socket-side receive window
+ * @sock_rcv_wup:	rcv_nxt on last window update sent
+ * @snd_wscale:		Window scaling factor, send
+ * @snd_wscale:		Window scaling factor, receive
+ */
+struct tcp_tap_transfer_ext {
+	uint32_t	sock_seq_snd;
+	uint32_t	sock_seq_rcv;
+
+	uint32_t	sndlen;
+	uint32_t	rcvlen;
+
+	uint32_t	sock_mss;
+
+	/* We can't just use struct tcp_repair_window: we need network order */
+	uint32_t	sock_snd_wl1;
+	uint32_t	sock_snd_wnd;
+	uint32_t	sock_max_window;
+	uint32_t	sock_rcv_wnd;
+	uint32_t	sock_rcv_wup;
+
+	uint8_t		snd_wscale;
+	uint8_t		rcv_wscale;
+} __attribute__((packed, aligned(__alignof__(uint32_t))));
+
 /**
  * struct tcp_splice_conn - Descriptor for a spliced TCP connection
  * @f:			Generic flow information
@@ -140,6 +221,12 @@ extern int init_sock_pool4	[TCP_SOCK_POOL_SIZE];
 extern int init_sock_pool6	[TCP_SOCK_POOL_SIZE];
 
 bool tcp_flow_defer(const struct tcp_tap_conn *conn);
+int tcp_flow_repair_on(struct ctx *c, const struct tcp_tap_conn *conn);
+int tcp_flow_repair_off(struct ctx *c, const struct tcp_tap_conn *conn);
+int tcp_flow_migrate_source(int fd, const struct tcp_tap_conn *conn);
+int tcp_flow_migrate_source_ext(int fd, const struct tcp_tap_conn *conn);
+int tcp_flow_migrate_target(struct ctx *c, int fd);
+int tcp_flow_migrate_target_ext(struct ctx *c, union flow *flow, int fd);
 bool tcp_splice_flow_defer(struct tcp_splice_conn *conn);
 void tcp_splice_timer(const struct ctx *c, struct tcp_splice_conn *conn);
 int tcp_conn_pool_sock(int pool[]);
-- 
@@ -96,6 +96,87 @@ struct tcp_tap_conn {
 	uint32_t	seq_init_from_tap;
 };
 
+/**
+ * struct tcp_tap_transfer - Migrated TCP data, flow table part, network order
+ * @pif:		Interfaces for each side of the flow
+ * @side:		Addresses and ports for each side of the flow
+ * @retrans:		Number of retransmissions occurred due to ACK_TIMEOUT
+ * @ws_from_tap:	Window scaling factor advertised from tap/guest
+ * @ws_to_tap:		Window scaling factor advertised to tap/guest
+ * @events:		Connection events, implying connection states
+ * @tap_mss:		MSS advertised by tap/guest, rounded to 2 ^ TCP_MSS_BITS
+ * @sndbuf:		Sending buffer in kernel, rounded to 2 ^ SNDBUF_BITS
+ * @flags:		Connection flags representing internal attributes
+ * @seq_dup_ack_approx:	Last duplicate ACK number sent to tap
+ * @wnd_from_tap:	Last window size from tap, unscaled (as received)
+ * @wnd_to_tap:		Sending window advertised to tap, unscaled (as sent)
+ * @seq_to_tap:		Next sequence for packets to tap
+ * @seq_ack_from_tap:	Last ACK number received from tap
+ * @seq_from_tap:	Next sequence for packets from tap (not actually sent)
+ * @seq_ack_to_tap:	Last ACK number sent to tap
+ * @seq_init_from_tap:	Initial sequence number from tap
+*/
+struct tcp_tap_transfer {
+	uint8_t		pif[SIDES];
+	struct flowside	side[SIDES];
+
+	uint8_t		retrans;
+	uint8_t		ws_from_tap;
+	uint8_t		ws_to_tap;
+	uint8_t		events;
+
+	uint32_t	tap_mss;
+
+	uint32_t	sndbuf;
+
+	uint8_t		flags;
+	uint8_t		seq_dup_ack_approx;
+
+	uint16_t	wnd_from_tap;
+	uint16_t	wnd_to_tap;
+
+	uint32_t	seq_to_tap;
+	uint32_t	seq_ack_from_tap;
+	uint32_t	seq_from_tap;
+	uint32_t	seq_ack_to_tap;
+	uint32_t	seq_init_from_tap;
+} __attribute__((packed, aligned(__alignof__(uint32_t))));
+
+/**
+ * struct tcp_tap_transfer_ext - Migrated TCP data, outside flow, network order
+ * @sock_seq_snd:	Socket-side send sequence
+ * @sock_seq_rcv:	Socket-side receive sequence
+ * @sndlen:		Length of pending send queue
+ * @rcvlen:		Length of pending receive queue
+ * @sock_mss:		Socket-side MSS
+ * @sock_snd_wl1:	Next sequence used in window probe (next sequence - 1)
+ * @sock_snd_wnd:	Socket-side sending window
+ * @sock_max_window:	Window clamp
+ * @sock_rcv_wnd:	Socket-side receive window
+ * @sock_rcv_wup:	rcv_nxt on last window update sent
+ * @snd_wscale:		Window scaling factor, send
+ * @snd_wscale:		Window scaling factor, receive
+ */
+struct tcp_tap_transfer_ext {
+	uint32_t	sock_seq_snd;
+	uint32_t	sock_seq_rcv;
+
+	uint32_t	sndlen;
+	uint32_t	rcvlen;
+
+	uint32_t	sock_mss;
+
+	/* We can't just use struct tcp_repair_window: we need network order */
+	uint32_t	sock_snd_wl1;
+	uint32_t	sock_snd_wnd;
+	uint32_t	sock_max_window;
+	uint32_t	sock_rcv_wnd;
+	uint32_t	sock_rcv_wup;
+
+	uint8_t		snd_wscale;
+	uint8_t		rcv_wscale;
+} __attribute__((packed, aligned(__alignof__(uint32_t))));
+
 /**
  * struct tcp_splice_conn - Descriptor for a spliced TCP connection
  * @f:			Generic flow information
@@ -140,6 +221,12 @@ extern int init_sock_pool4	[TCP_SOCK_POOL_SIZE];
 extern int init_sock_pool6	[TCP_SOCK_POOL_SIZE];
 
 bool tcp_flow_defer(const struct tcp_tap_conn *conn);
+int tcp_flow_repair_on(struct ctx *c, const struct tcp_tap_conn *conn);
+int tcp_flow_repair_off(struct ctx *c, const struct tcp_tap_conn *conn);
+int tcp_flow_migrate_source(int fd, const struct tcp_tap_conn *conn);
+int tcp_flow_migrate_source_ext(int fd, const struct tcp_tap_conn *conn);
+int tcp_flow_migrate_target(struct ctx *c, int fd);
+int tcp_flow_migrate_target_ext(struct ctx *c, union flow *flow, int fd);
 bool tcp_splice_flow_defer(struct tcp_splice_conn *conn);
 void tcp_splice_timer(const struct ctx *c, struct tcp_splice_conn *conn);
 int tcp_conn_pool_sock(int pool[]);
-- 
2.43.0