introduce rx and tx session modes

And convert existing --msg-trunc and --msg-zerocopy to this
new message.

Potentially io_uring can be a separate mode as well.

Signed-off-by: Stanislav Fomichev <[email protected]>

Author: Stanislav Fomichev

client.c

@@ -532,6 +532,8 @@ dump_result_machine(struct kpm_test_results *result, const char *dir,
 
 int main(int argc, char *argv[])
 {
+	enum kpm_rx_mode rx_mode = KPM_RX_MODE_SOCKET;
+	enum kpm_tx_mode tx_mode = KPM_TX_MODE_SOCKET;
 	unsigned int src_ncpus, dst_ncpus;
 	struct __kpm_generic_u32 *ack_id;
 	__u32 *src_wrk_cpu, *dst_wrk_cpu;
@@ -607,6 +609,22 @@ int main(int argc, char *argv[])
 		goto out;
 	}
 
+	if (opt.msg_trunc)
+		rx_mode = KPM_RX_MODE_SOCKET_TRUNC;
+
+	if (opt.msg_zerocopy)
+		tx_mode = KPM_TX_MODE_SOCKET_ZEROCOPY;
+
+	if (kpm_req_mode(dst, rx_mode, tx_mode) < 0) {
+		warnx("Failed setup destination mode");
+		goto out;
+	}
+
+	if (kpm_req_mode(src, rx_mode, tx_mode) < 0) {
+		warnx("Failed setup source mode");
+		goto out;
+	}
+
 	conns = spawn_conn(src, dst, &conn_addr, len);
 	if (!conns)
 		goto out;
@@ -677,8 +695,6 @@ int main(int argc, char *argv[])
 		test->specs[i].worker_id = dst_wrk_id[i];
 		test->specs[i].read_size = opt.read_size;
 		test->specs[i].write_size = opt.write_size;
-		test->specs[i].msg_trunc = opt.msg_trunc;
-		test->specs[i].msg_zerocopy = opt.msg_zerocopy;
 		if (opt.req_size == ~0U) {
 			test->specs[i].type = KPM_TEST_TYPE_STREAM;
 		} else {

proto.c

@@ -180,6 +180,16 @@ int kpm_send_tcp_cc(int fd, __u32 id, char *cc_name)
 	return kpm_send(fd, &msg.hdr, sizeof(msg), KPM_MSG_TYPE_TCP_CC);
 }
 
+int kpm_send_mode(int fd, enum kpm_rx_mode rx_mode, enum kpm_tx_mode tx_mode)
+{
+	struct kpm_mode msg = {};
+
+	msg.rx_mode = rx_mode;
+	msg.tx_mode = tx_mode;
+
+	return kpm_send(fd, &msg.hdr, sizeof(msg), KPM_MSG_TYPE_MODE);
+}
+
 int kpm_send_pin_worker(int fd, __u32 id, __u32 cpu)
 {
 	struct kpm_pin_worker msg;
@@ -443,6 +453,34 @@ kpm_req_tcp_cc(int fd, __u32 conn_id, char *cc_name)
 	return 0;
 }
 
+int
+kpm_req_mode(int fd, enum kpm_rx_mode rx_mode, enum kpm_tx_mode tx_mode)
+{
+	struct kpm_empty *repl;
+	int id;
+
+	id = kpm_send_mode(fd, rx_mode, tx_mode);
+	if (id < 0) {
+		warnx("Failed to request mode");
+		return id;
+	}
+
+	repl = kpm_receive(fd);
+	if (!repl) {
+		warnx("Failed to request mode - no response");
+		return -1;
+	}
+
+	if (!kpm_good_reply(repl, KPM_MSG_TYPE_MODE, id)) {
+		warnx("Failed to request mode - bad reply");
+		free(repl);
+		return -1;
+	}
+
+	free(repl);
+	return 0;
+}
+
 int kpm_req_disconnect(int fd, __u32 connection_id)
 {
 	struct kpm_empty *repl;

proto.h

@@ -25,6 +25,7 @@ enum kpm_msg_type {
 	KPM_MSG_TYPE_TLS,
 	KPM_MSG_TYPE_MAX_PACING,
 	KPM_MSG_TYPE_TCP_CC,
+	KPM_MSG_TYPE_MODE,
 	KPM_MSG_TYPE_TEST,
 	KPM_MSG_TYPE_TEST_RESULT,
 	KPM_MSG_TYPE_END_TEST,
@@ -121,6 +122,22 @@ struct kpm_tcp_cc {
 	char cc_name[KPM_CC_NAME_LEN];
 };
 
+enum kpm_rx_mode {
+	KPM_RX_MODE_SOCKET,
+	KPM_RX_MODE_SOCKET_TRUNC,
+};
+
+enum kpm_tx_mode {
+	KPM_TX_MODE_SOCKET,
+	KPM_TX_MODE_SOCKET_ZEROCOPY,
+};
+
+struct kpm_mode {
+	struct kpm_header hdr;
+	enum kpm_rx_mode rx_mode;
+	enum kpm_tx_mode tx_mode;
+};
+
 enum kpm_tls_mask {
 	KPM_TLS_ULP = 1,
 	KPM_TLS_TX = 2,
@@ -164,8 +181,6 @@ struct kpm_test {
 		enum kpm_test_type type;
 		__u32 read_size;
 		__u32 write_size;
-		__u32 msg_trunc:1;
-		__u32 msg_zerocopy:1;
 		union kpm_test_arg {
 			struct {
 				__u32 req_size;
@@ -254,6 +269,7 @@ int kpm_send_tls(int fd, __u32 conn_id, __u32 dir_mask,
 		 void *info, socklen_t len);
 int kpm_send_max_pacing(int fd, __u32 id, __u32 max_pace);
 int kpm_send_tcp_cc(int fd, __u32 id, char *cc_name);
+int kpm_send_mode(int fd, enum kpm_rx_mode rx_mode, enum kpm_tx_mode tx_mode);
 int kpm_send_pin_worker(int fd, __u32 id, __u32 cpu);
 
 void kpm_reply_error(int fd, struct kpm_header *hdr, __u16 error);
@@ -276,6 +292,7 @@ int kpm_req_tls(int fd, __u32 conn_id, __u32 dir_mask,
 		void *info, socklen_t len);
 int kpm_req_pacing(int fd, __u32 conn_id, __u32 max_pace);
 int kpm_req_tcp_cc(int fd, __u32 conn_id, char *cc_name);
+int kpm_req_mode(int fd, enum kpm_rx_mode rx_mode, enum kpm_tx_mode tx_mode);
 int kpm_req_disconnect(int fd, __u32 connection_id);
 
 #endif /* PROTO_H */

server.h

@@ -10,6 +10,8 @@
 #include <ccan/compiler/compiler.h>
 #include <ccan/list/list.h>
 
+#include "proto.h"
+
 struct server_session {
 	int cfd;
 	pid_t pid;
@@ -19,6 +21,6 @@ struct server_session {
 struct server_session *
 server_session_spawn(int fd, struct sockaddr_in6 *addr, socklen_t *addrlen);
 
-void NORETURN pworker_main(int fd);
+void NORETURN pworker_main(int fd, enum kpm_rx_mode rx_mode, enum kpm_tx_mode tx_mode);
 
 #endif /* SERVER_H */

server_session.c

@@ -33,6 +33,8 @@ struct session_state {
 	int epollfd;
 	int quit;
 	int tcp_sock;
+	enum kpm_rx_mode rx_mode;
+	enum kpm_tx_mode tx_mode;
 	unsigned int connection_ids;
 	unsigned int worker_ids;
 	unsigned int test_ids;
@@ -495,6 +497,31 @@ server_msg_tcp_cc(struct session_state *self, struct kpm_header *hdr)
 	self->quit = 1;
 }
 
+static void
+server_msg_mode(struct session_state *self, struct kpm_header *hdr)
+{
+	struct kpm_mode *req;
+
+	if (hdr->len < sizeof(*req)) {
+		warn("Invalid request in %s", __func__);
+		goto err_quit;
+	}
+	req = (void *)hdr;
+
+	self->rx_mode = req->rx_mode;
+	self->tx_mode = req->tx_mode;
+
+	if (kpm_reply_empty(self->main_sock, hdr) < 1) {
+		warnx("Reply failed");
+		goto err_quit;
+	}
+
+	return;
+
+err_quit:
+	self->quit = 1;
+}
+
 static void
 server_msg_spawn_pworker(struct session_state *self, struct kpm_header *hdr)
 {
@@ -519,7 +546,7 @@ server_msg_spawn_pworker(struct session_state *self, struct kpm_header *hdr)
 	}
 	if (!pwrk->pid) {
 		close(p[0]);
-		pworker_main(p[1]);
+		pworker_main(p[1], self->rx_mode, self->tx_mode);
 		exit(1);
 	}
 
@@ -788,6 +815,9 @@ static void session_handle_main_sock(struct session_state *self)
 	case KPM_MSG_TYPE_TCP_CC:
 		server_msg_tcp_cc(self, hdr);
 		break;
+	case KPM_MSG_TYPE_MODE:
+		server_msg_mode(self, hdr);
+		break;
 	case KPM_MSG_TYPE_SPAWN_PWORKER:
 		server_msg_spawn_pworker(self, hdr);
 		break;

worker.c

@@ -28,6 +28,8 @@
 /* Main worker state AKA self */
 struct worker_state {
 	int main_sock;
+	enum kpm_rx_mode rx_mode;
+	enum kpm_tx_mode tx_mode;
 	int epollfd;
 	unsigned int id;
 	int quit;
@@ -319,7 +321,7 @@ worker_msg_test(struct worker_state *self, struct kpm_header *hdr)
 		else
 			conn->to_recv = len;
 
-		zc = !!conn->spec->msg_zerocopy;
+		zc = self->tx_mode == KPM_TX_MODE_SOCKET_ZEROCOPY;
 		if (setsockopt(conn->fd, SOL_SOCKET, SO_ZEROCOPY, &zc, sizeof(zc))) {
 			warnx("Failed to set SO_ZEROCOPY");
 			self->quit = 1;
@@ -570,7 +572,8 @@ worker_handle_send(struct worker_state *self, struct connection *conn,
 		   unsigned int events)
 {
 	unsigned int rep = max_t(int, 10, conn->to_send / conn->write_size + 1);
-	int flags = conn->spec->msg_zerocopy ? MSG_ZEROCOPY : 0;
+	bool msg_zerocopy = self->tx_mode == KPM_TX_MODE_SOCKET_ZEROCOPY;
+	int flags = msg_zerocopy ? MSG_ZEROCOPY : 0;
 
 	while (rep--) {
 		void *src = &patbuf[conn->tot_sent % PATTERN_PERIOD];
@@ -598,7 +601,7 @@ worker_handle_send(struct worker_state *self, struct connection *conn,
 
 		conn->to_send -= n;
 		conn->tot_sent += n;
-		if (conn->spec->msg_zerocopy) {
+		if (msg_zerocopy) {
 			conn->to_send_comp += 1;
 			kpm_dbg("queued send completion, total %d",
 				conn->to_send_comp);
@@ -619,13 +622,14 @@ worker_handle_send(struct worker_state *self, struct connection *conn,
 
 static ssize_t worker_handle_regular_recv(struct worker_state *self, struct connection *conn, size_t chunk, int rep)
 {
+	bool msg_trunc = self->rx_mode == KPM_RX_MODE_SOCKET_TRUNC;
 	void *src = &patbuf[conn->tot_recv % PATTERN_PERIOD];
-	int flags = conn->spec->msg_trunc ? MSG_TRUNC : 0;
+	int flags = msg_trunc ? MSG_TRUNC : 0;
 	ssize_t n;
 
 	n = recv(conn->fd, conn->rxbuf, chunk, MSG_DONTWAIT | flags);
 
-	if (n <= 0 || conn->spec->msg_trunc)
+	if (n <= 0 || msg_trunc)
 		return n;
 
 	if (memcmp(conn->rxbuf, src, n))
@@ -709,9 +713,13 @@ worker_handle_conn(struct worker_state *self, int fd, unsigned int events)
 
 /* == Main loop == */
 
-void NORETURN pworker_main(int fd)
+void NORETURN pworker_main(int fd, enum kpm_rx_mode rx_mode, enum kpm_tx_mode tx_mode)
 {
-	struct worker_state self = { .main_sock = fd, };
+	struct worker_state self = {
+		.main_sock = fd,
+		.rx_mode = rx_mode,
+		.tx_mode = tx_mode,
+	};
 	struct epoll_event ev, events[32];
 	unsigned char j;
 	int i, nfds;