[PATCH v4 6/8] flow, tcp: Basic pre-migration source handler to dump sequence numbers

[Stefano Brivio <sbrivio@redhat.com>]

Very much draft quality, but it works. Ask passt-repair to switch
TCP sockets to repair mode and dump their current sequence numbers to
the flow table, which will be transferred and used by the target in
the next step.

Signed-off-by: Stefano Brivio <sbrivio@redhat.com>
---
 flow.c       | 49 +++++++++++++++++++++++++++++++++++++++++++++
 flow.h       |  2 ++
 migrate.c    | 15 +++++++-------
 tcp.c        | 56 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 tcp_conn.h   |  5 +++++
 vhost_user.c |  1 +
 6 files changed, 121 insertions(+), 7 deletions(-)

diff --git a/flow.c b/flow.c
index 3b8dd0e..da738eb 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",
@@ -870,6 +871,54 @@ void flow_defer_handler(const struct ctx *c, const struct timespec *now)
 	*last_next = FLOW_MAX;
 }
 
+/**
+ * flow_migrate_source_pre() - Prepare all source flows for migration
+ * @c:		Execution context
+ * @m:		Migration metadata
+ * @stage:	Migration stage information (unused)
+ * @fd:		Migration fd (unused)
+ *
+ * Return: 0 on success
+ */
+int flow_migrate_source_pre(struct ctx *c, struct migrate_meta *m,
+			    const struct migrate_stage *stage, int fd)
+{
+	unsigned i;
+	int rc;
+
+	(void)m;
+	(void)stage;
+	(void)fd;
+
+	for (i = 0; i < FLOW_MAX; i++) {	/* TODO: iterator with skip */
+		union flow *flow = &flowtab[i];
+
+		if (flow->f.state == FLOW_STATE_FREE)
+			i += flow->free.n - 1;
+		else if (flow->f.state == FLOW_STATE_ACTIVE &&
+			 flow->f.type == FLOW_TCP)
+			rc = tcp_flow_repair_on(c, &flow->tcp);
+
+		if (rc)
+			return rc;		/* TODO: rollback */
+	}
+
+	if ((rc = repair_flush(c)))		/* TODO: move to TCP logic */
+		return rc;
+
+	for (i = 0; i < FLOW_MAX; i++) {	/* TODO: iterator with skip */
+		union flow *flow = &flowtab[i];
+
+		if (flow->f.state == FLOW_STATE_FREE)
+			i += flow->free.n - 1;
+		else if (flow->f.state == FLOW_STATE_ACTIVE &&
+			 flow->f.type == FLOW_TCP)
+			tcp_flow_dump_seq(c, &flow->tcp);
+	}
+
+	return 0;
+}
+
 /**
  * flow_init() - Initialise flow related data structures
  */
diff --git a/flow.h b/flow.h
index 24ba3ef..c526938 100644
--- a/flow.h
+++ b/flow.h
@@ -249,6 +249,8 @@ 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, struct migrate_meta *m,
+			    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 f64ffc8..b42b12f 100644
--- a/migrate.c
+++ b/migrate.c
@@ -21,9 +21,9 @@
 #include "inany.h"
 #include "flow.h"
 #include "flow_table.h"
-#include "repair.h"
 
 #include "migrate.h"
+#include "repair.h"
 
 /* Current version of migration data */
 #define MIGRATE_VERSION		1
@@ -104,6 +104,7 @@ static int migrate_recv_block(struct ctx *c, struct migrate_meta *m,
 static const struct migrate_stage stages_v1[] = {
 	{
 		.name = "flow pre",
+		.source = flow_migrate_source_pre,
 		.target = NULL,
 	},
 	DATA_STAGE(flow_first_free),
@@ -129,7 +130,7 @@ static const struct migrate_version versions[] = {
  *
  * Return: 0 on success, positive error code on failure
  */
-static int migrate_source(struct ctx *c, int fd)
+int migrate_source(struct ctx *c, int fd)
 {
 	struct migrate_meta m;
 	unsigned i;
@@ -220,7 +221,7 @@ static int migrate_target_read_header(int fd, struct migrate_meta *m)
  *
  * Return: 0 on success, positive error code on failure
  */
-static int migrate_target(struct ctx *c, int fd)
+int migrate_target(struct ctx *c, int fd)
 {
 	struct migrate_meta m;
 	unsigned i;
@@ -258,6 +259,7 @@ void migrate_init(struct ctx *c)
 {
 	c->device_state_fd = -1;
 	c->device_state_result = -1;
+	repair_sock_init(c);
 }
 
 /**
@@ -268,6 +270,7 @@ void migrate_close(struct ctx *c)
 {
 	if (c->device_state_fd != -1) {
 		debug("Closing migration channel, fd: %d", c->device_state_fd);
+		epoll_del(c, c->device_state_fd);
 		close(c->device_state_fd);
 		c->device_state_fd = -1;
 		c->device_state_result = -1;
@@ -282,14 +285,12 @@ void migrate_close(struct ctx *c)
  */
 void migrate_request(struct ctx *c, int fd, bool target)
 {
-	debug("Migration requested, fd: %d (was %d)",
-	      fd, c->device_state_fd);
+	debug("Migration requested, fd: %d", c->device_state_fd);
 
 	if (c->device_state_fd != -1)
 		migrate_close(c);
 
 	c->device_state_fd = fd;
-	c->migrate_target = target;
 }
 
 /**
@@ -304,7 +305,7 @@ void migrate_handler(struct ctx *c)
 		return;
 
 	debug("migrate_handler fd %d target %d",
-	      c->device_state_fd, c->migrate_target);
+		c->device_state_fd, c->migrate_target);
 
 	if (c->migrate_target)
 		rc = migrate_target(c, c->device_state_fd);
diff --git a/tcp.c b/tcp.c
index fac322c..96d7649 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"
@@ -868,6 +869,61 @@ void tcp_defer_handler(struct ctx *c)
 	tcp_payload_flush(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_dump_seq() - Dump sequences for send and receive queues
+ * @c:		Execution context
+ * @conn:	Pointer to the TCP connection structure
+ *
+ * Return: 0 on success, negative error code on failure
+ */
+int tcp_flow_dump_seq(struct ctx *c, struct tcp_tap_conn *conn)
+{
+	int v, s = conn->sock;
+	socklen_t vlen;
+
+	(void)c;
+
+	vlen = sizeof(v);
+
+	v = TCP_SEND_QUEUE;
+	/* TODO: proper error management and prints */
+	if (setsockopt(s, SOL_TCP, TCP_REPAIR_QUEUE, &v, vlen))
+		return -errno;
+
+	if (getsockopt(s, SOL_TCP, TCP_QUEUE_SEQ, &conn->sock_seq_snd, &vlen))
+		return -errno;
+
+	debug("Send queue sequence %u for socket %i", conn->sock_seq_snd, s);
+
+	v = TCP_RECV_QUEUE;
+	if (setsockopt(s, SOL_TCP, TCP_REPAIR_QUEUE, &v, vlen))
+		return -errno;
+
+	if (getsockopt(s, SOL_TCP, TCP_QUEUE_SEQ, &conn->sock_seq_rcv, &vlen))
+		return -errno;
+
+	debug("Receive queue sequence %u for socket %i", conn->sock_seq_rcv, s);
+
+	return 0;
+}
+
 /**
  * tcp_fill_header() - Fill the TCP header fields for a given TCP segment.
  *
diff --git a/tcp_conn.h b/tcp_conn.h
index d342680..0c3e197 100644
--- a/tcp_conn.h
+++ b/tcp_conn.h
@@ -94,6 +94,9 @@ struct tcp_tap_conn {
 	uint32_t	seq_from_tap;
 	uint32_t	seq_ack_to_tap;
 	uint32_t	seq_init_from_tap;
+
+	uint32_t	sock_seq_snd;
+	uint32_t	sock_seq_rcv;
 };
 
 /**
@@ -140,6 +143,8 @@ 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_dump_seq(struct ctx *c, struct tcp_tap_conn *conn);
 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[]);
diff --git a/vhost_user.c b/vhost_user.c
index b107d0f..63f745c 100644
--- a/vhost_user.c
+++ b/vhost_user.c
@@ -44,6 +44,7 @@
 #include "tap.h"
 #include "vhost_user.h"
 #include "pcap.h"
+#include "migrate.h"
 
 /* vhost-user version we are compatible with */
 #define VHOST_USER_VERSION 1
-- 
@@ -44,6 +44,7 @@
 #include "tap.h"
 #include "vhost_user.h"
 #include "pcap.h"
+#include "migrate.h"
 
 /* vhost-user version we are compatible with */
 #define VHOST_USER_VERSION 1
-- 
2.43.0