[SPDK] [RFC PATCH 4/4] rpc: add blocked version of rpc server with epoll

[SPDK] [RFC PATCH 4/4] rpc: add blocked version of rpc server with epoll

[Ming Lin]

[-- Attachment #1: Type: text/plain, Size: 5647 bytes --]

From: Ming Lin <ming.lin(a)alibaba-inc.com>

Instead of busy polling, rpc server sleeps if nothing to handle.

Signed-off-by: Ming Lin <ming.lin(a)alibaba-inc.com>
---
 include/spdk/jsonrpc.h           |  4 ++++
 include/spdk/rpc.h               |  1 +
 lib/jsonrpc/jsonrpc_internal.h   |  1 +
 lib/jsonrpc/jsonrpc_server_tcp.c | 36 ++++++++++++++++++++++++++++--
 lib/rpc/rpc.c                    | 47 ++++++++++++++++++++++++++++++++++++++++
 5 files changed, 87 insertions(+), 2 deletions(-)

diff --git a/include/spdk/jsonrpc.h b/include/spdk/jsonrpc.h
index 0ca85ea..f429b1e 100644
--- a/include/spdk/jsonrpc.h
+++ b/include/spdk/jsonrpc.h
@@ -74,6 +74,10 @@ int spdk_jsonrpc_server_poll(struct spdk_jsonrpc_server *server);
 
 void spdk_jsonrpc_server_shutdown(struct spdk_jsonrpc_server *server);
 
+int spdk_jsonrpc_server_sockfd(struct spdk_jsonrpc_server *server);
+
+void spdk_jsonrpc_server_set_efd(struct spdk_jsonrpc_server *server, int efd);
+
 /**
  * Begin building a response to a JSON-RPC request.
  *
diff --git a/include/spdk/rpc.h b/include/spdk/rpc.h
index 1ef3066..fe932dc 100644
--- a/include/spdk/rpc.h
+++ b/include/spdk/rpc.h
@@ -44,6 +44,7 @@ extern "C" {
 
 int spdk_rpc_listen(const char *listen_addr);
 void spdk_rpc_accept(void);
+void spdk_rpc_accept_epoll(void);
 void spdk_rpc_close(void);
 
 typedef void (*spdk_rpc_method_handler)(struct spdk_jsonrpc_request *request,
diff --git a/lib/jsonrpc/jsonrpc_internal.h b/lib/jsonrpc/jsonrpc_internal.h
index 5cafc3e..1657724 100644
--- a/lib/jsonrpc/jsonrpc_internal.h
+++ b/lib/jsonrpc/jsonrpc_internal.h
@@ -73,6 +73,7 @@ struct spdk_jsonrpc_server_conn {
 
 struct spdk_jsonrpc_server {
 	int sockfd;
+	int efd;
 	spdk_jsonrpc_handle_request_fn handle_request;
 	struct spdk_jsonrpc_server_conn conns[SPDK_JSONRPC_MAX_CONNS];
 	int num_conns;
diff --git a/lib/jsonrpc/jsonrpc_server_tcp.c b/lib/jsonrpc/jsonrpc_server_tcp.c
index 3a44ec8..8dc5885 100644
--- a/lib/jsonrpc/jsonrpc_server_tcp.c
+++ b/lib/jsonrpc/jsonrpc_server_tcp.c
@@ -33,6 +33,7 @@
 
 #include "jsonrpc_internal.h"
 #include "spdk/string.h"
+#include <sys/epoll.h>
 
 struct spdk_jsonrpc_server *
 spdk_jsonrpc_server_listen(int domain, int protocol,
@@ -103,6 +104,9 @@ spdk_jsonrpc_server_shutdown(struct spdk_jsonrpc_server *server)
 		close(server->conns[i].sockfd);
 	}
 
+	if (server->efd > 0)
+		close(server->efd);
+
 	free(server);
 }
 
@@ -112,6 +116,10 @@ spdk_jsonrpc_server_conn_close(struct spdk_jsonrpc_server_conn *conn)
 	conn->closed = true;
 
 	if (conn->sockfd >= 0) {
+		if (conn->server->efd > 0) {
+			epoll_ctl(conn->server->efd, EPOLL_CTL_DEL,
+				  conn->sockfd, NULL);
+		}
 		close(conn->sockfd);
 		conn->sockfd = -1;
 	}
@@ -165,6 +173,20 @@ spdk_jsonrpc_server_accept(struct spdk_jsonrpc_server *server)
 			return -1;
 		}
 
+		if (server->efd > 0) {
+			struct epoll_event event;
+			int r;
+
+			event.data.fd = conn->sockfd;
+			event.events = EPOLLIN | EPOLLET;
+			r = epoll_ctl(server->efd, EPOLL_CTL_ADD, conn->sockfd, &event);
+			if (r < 0) {
+				spdk_strerror_r(errno, buf, sizeof(buf));
+				SPDK_ERRLOG("epoll_ctl error for socket, fd: %d (%s)\n", conn->sockfd, buf);
+				return -1;
+			}
+		}
+
 		server->num_conns++;
 
 		return 0;
@@ -368,13 +390,13 @@ spdk_jsonrpc_server_poll(struct spdk_jsonrpc_server *server)
 			continue;
 		}
 
-		rc = spdk_jsonrpc_server_conn_send(conn);
+		rc = spdk_jsonrpc_server_conn_recv(conn);
 		if (rc != 0) {
 			spdk_jsonrpc_server_conn_close(conn);
 			continue;
 		}
 
-		rc = spdk_jsonrpc_server_conn_recv(conn);
+		rc = spdk_jsonrpc_server_conn_send(conn);
 		if (rc != 0) {
 			spdk_jsonrpc_server_conn_close(conn);
 			continue;
@@ -383,3 +405,13 @@ spdk_jsonrpc_server_poll(struct spdk_jsonrpc_server *server)
 
 	return 0;
 }
+
+int spdk_jsonrpc_server_sockfd(struct spdk_jsonrpc_server *server)
+{
+	return server->sockfd;
+}
+
+void spdk_jsonrpc_server_set_efd(struct spdk_jsonrpc_server *server, int efd)
+{
+	server->efd = efd;
+}
diff --git a/lib/rpc/rpc.c b/lib/rpc/rpc.c
index 55102f9..c467718 100644
--- a/lib/rpc/rpc.c
+++ b/lib/rpc/rpc.c
@@ -39,6 +39,8 @@
 #include "spdk/log.h"
 #include "spdk/string.h"
 
+#include <sys/epoll.h>
+
 #define RPC_DEFAULT_PORT	"5260"
 
 static struct sockaddr_un g_rpc_listen_addr_unix = {};
@@ -155,6 +157,51 @@ spdk_rpc_accept(void)
 	spdk_jsonrpc_server_poll(g_jsonrpc_server);
 }
 
+#define MAXEVENTS 64
+
+void
+spdk_rpc_accept_epoll(void)
+{
+	int efd, sfd;
+	struct epoll_event event;
+	struct epoll_event *events;
+	int ret;
+
+	efd = epoll_create1(0);
+	if (efd == -1) {
+		SPDK_ERRLOG("epoll_create fail\n");
+		return;
+	}
+
+	sfd = spdk_jsonrpc_server_sockfd(g_jsonrpc_server);
+	event.data.fd = sfd;
+	event.events = EPOLLIN | EPOLLET;
+
+	ret = epoll_ctl(efd, EPOLL_CTL_ADD, sfd, &event);
+	if (ret == -1) {
+		SPDK_ERRLOG("epoll_ctl fail\n");
+		close(efd);
+		return;
+	}
+
+	spdk_jsonrpc_server_set_efd(g_jsonrpc_server, efd);
+	events = calloc(MAXEVENTS, sizeof(event));
+
+	while (1) {
+		int n, i;
+
+		n = epoll_wait(efd, events, MAXEVENTS, -1);
+		for (i = 0; i < n; i++) {
+			/*
+			 * Always call spdk_jsonrpc_server_poll() even the polling fd
+			 * returns error which means client side has closed the socket.
+			 * spdk_jsonrpc_server_poll() will cleanup it.
+			 */
+			spdk_jsonrpc_server_poll(g_jsonrpc_server);
+		}
+	}
+}
+
 void
 spdk_rpc_register_method(const char *method, spdk_rpc_method_handler func)
 {
-- 
1.9.1