H2O: Remove the write polling work-around (#2705)

libh2o's event loop now supports write polling without sending
any data, so the current work-around is no longer necessary.

frameworks/C/h2o/src/database.c

@@ -37,7 +37,6 @@ typedef struct {
 	list_t l;
 	PGconn *conn;
 	thread_context_t *ctx;
-	void (*on_write_ready)(void *);
 	db_query_param_t *param;
 	h2o_socket_t *sock;
 	size_t prep_stmt_idx;
@@ -53,12 +52,10 @@ static void on_database_connect_timeout(h2o_timeout_entry_t *entry);
 static void on_database_error(db_conn_t *db_conn, const char *error_string);
 static void on_database_read_ready(h2o_socket_t *db_sock, const char *err);
 static void on_database_timeout(h2o_timeout_entry_t *entry);
-static void on_database_write_ready(void *data);
+static void on_database_write_ready(h2o_socket_t *db_sock, const char *err);
 static void poll_database_connection(h2o_socket_t *db_sock, const char *err);
 static void process_query(db_conn_t *db_conn);
 static void start_database_connect(thread_context_t *ctx, db_conn_t *db_conn);
-static int start_database_write_polling(db_conn_t *db_conn);
-static void stop_database_write_polling(db_conn_t *db_conn);
 
 static const struct {
 	const char *name;
@@ -80,10 +77,10 @@ static int do_database_write(db_conn_t *db_conn)
 		ERROR(PQerrorMessage(db_conn->conn));
 		on_database_error(db_conn, DB_ERROR);
 	}
-	else if (start_database_write_polling(db_conn))
-		on_database_error(db_conn, DB_ERROR);
-	else
+	else {
+		h2o_socket_notify_write(db_conn->sock, on_database_write_ready);
 		ret = 0;
+	}
 
 	return ret;
 }
@@ -109,7 +106,7 @@ static void do_execute_query(db_conn_t *db_conn)
 		                 &db_conn->ctx->db_state.h2o_timeout,
 		                 &db_conn->h2o_timeout_entry);
 		h2o_socket_read_start(db_conn->sock, on_database_read_ready);
-		on_database_write_ready(&db_conn->on_write_ready);
+		on_database_write_ready(db_conn->sock, NULL);
 	}
 	else {
 		ERROR(PQerrorMessage(db_conn->conn));
@@ -192,10 +189,8 @@ static void on_database_read_ready(h2o_socket_t *db_sock, const char *err)
 		if (PQconsumeInput(db_conn->conn)) {
 			const int send_status = PQflush(db_conn->conn);
 
-			if (send_status > 0 && start_database_write_polling(db_conn)) {
-				on_database_error(db_conn, DB_ERROR);
-				return;
-			}
+			if (send_status > 0)
+				h2o_socket_notify_write(db_conn->sock, on_database_write_ready);
 
 			if (send_status >= 0) {
 				while (!PQisBusy(db_conn->conn)) {
@@ -257,26 +252,32 @@ static void on_database_timeout(h2o_timeout_entry_t *entry)
 	start_database_connect(db_conn->ctx, db_conn);
 }
 
-static void on_database_write_ready(void *data)
+static void on_database_write_ready(h2o_socket_t *db_sock, const char *err)
 {
-	db_conn_t * const db_conn = H2O_STRUCT_FROM_MEMBER(db_conn_t, on_write_ready, data);
+	db_conn_t * const db_conn = db_sock->data;
 
-	if (db_conn->prep_stmt_idx) {
-		const int send_status = PQflush(db_conn->conn);
+	if (err) {
+		ERROR(err);
+		on_database_error(db_conn, DB_ERROR);
+	}
+	else {
+		if (db_conn->prep_stmt_idx) {
+			const int send_status = PQflush(db_conn->conn);
 
-		if (!send_status) {
-			if (db_conn->flags & IS_WRITING && db_conn->param)
-				do_database_write(db_conn);
-		}
-		else if (send_status < 0) {
-			LIBRARY_ERROR("PQflush", PQerrorMessage(db_conn->conn));
-			on_database_error(db_conn, DB_ERROR);
+			if (!send_status) {
+				if (db_conn->flags & IS_WRITING && db_conn->param)
+					do_database_write(db_conn);
+			}
+			else if (send_status < 0) {
+				LIBRARY_ERROR("PQflush", PQerrorMessage(db_conn->conn));
+				on_database_error(db_conn, DB_ERROR);
+			}
+			else
+				h2o_socket_notify_write(db_conn->sock, on_database_write_ready);
 		}
-		else if (send_status > 0 && start_database_write_polling(db_conn))
-			on_database_error(db_conn, DB_ERROR);
+		else
+			poll_database_connection(db_conn->sock, NULL);
 	}
-	else
-		poll_database_connection(db_conn->sock, NULL);
 }
 
 static void poll_database_connection(h2o_socket_t *db_sock, const char *err)
@@ -292,9 +293,7 @@ static void poll_database_connection(h2o_socket_t *db_sock, const char *err)
 
 		switch (status) {
 			case PGRES_POLLING_WRITING:
-				if (start_database_write_polling(db_conn))
-					break;
-
+				h2o_socket_notify_write(db_conn->sock, on_database_write_ready);
 				return;
 			case PGRES_POLLING_OK:
 				if (PQsetnonblocking(db_conn->conn, 1)) {
@@ -330,7 +329,7 @@ static void process_query(db_conn_t *db_conn)
 			                 &db_conn->ctx->db_state.h2o_timeout,
 			                 &db_conn->h2o_timeout_entry);
 			h2o_socket_read_start(db_conn->sock, on_database_read_ready);
-			on_database_write_ready(&db_conn->on_write_ready);
+			on_database_write_ready(db_conn->sock, NULL);
 		}
 		else {
 			LIBRARY_ERROR("PQsendPrepare", PQerrorMessage(db_conn->conn));
@@ -364,7 +363,6 @@ static void start_database_connect(thread_context_t *ctx, db_conn_t *db_conn)
 		db_conn->prep_stmt_idx = 0;
 		db_conn->flags = IS_RESETTING;
 		h2o_timeout_unlink(&db_conn->h2o_timeout_entry);
-		stop_database_write_polling(db_conn);
 		h2o_socket_read_stop(db_conn->sock);
 		h2o_socket_close(db_conn->sock);
 
@@ -424,13 +422,8 @@ static void start_database_connect(thread_context_t *ctx, db_conn_t *db_conn)
 		                 &ctx->db_state.h2o_timeout,
 		                 &db_conn->h2o_timeout_entry);
 		h2o_socket_read_start(db_conn->sock, poll_database_connection);
-
-		if (!start_database_write_polling(db_conn))
-			return;
-
-		h2o_socket_read_stop(db_conn->sock);
-		h2o_timeout_unlink(&db_conn->h2o_timeout_entry);
-		h2o_socket_close(db_conn->sock);
+		h2o_socket_notify_write(db_conn->sock, on_database_write_ready);
+		return;
 	}
 	else {
 		errno = ENOMEM;
@@ -446,23 +439,6 @@ error:
 	error_notification(ctx, false, DB_ERROR);
 }
 
-static int start_database_write_polling(db_conn_t *db_conn)
-{
-	const bool rearm = !!db_conn->on_write_ready;
-
-	db_conn->on_write_ready = on_database_write_ready;
-	return start_write_polling(PQsocket(db_conn->conn),
-	                           &db_conn->on_write_ready,
-	                           rearm,
-	                           &db_conn->ctx->event_loop);
-}
-
-static void stop_database_write_polling(db_conn_t *db_conn)
-{
-	db_conn->on_write_ready = NULL;
-	stop_write_polling(PQsocket(db_conn->conn), &db_conn->ctx->event_loop);
-}
-
 void connect_to_database(thread_context_t *ctx)
 {
 	for (size_t i = ctx->config->max_db_conn_num - ctx->db_state.db_conn_num; i > 0; i--)

frameworks/C/h2o/src/event_loop.c

@@ -32,7 +32,6 @@
 #include <netinet/in.h>
 #include <netinet/tcp.h>
 #include <openssl/ssl.h>
-#include <sys/epoll.h>
 #include <sys/socket.h>
 #include <sys/types.h>
 
@@ -42,11 +41,9 @@
 #include "utility.h"
 
 #define DEFAULT_TCP_FASTOPEN_QUEUE_LEN 4096
-#define MAX_EPOLL_EVENTS 64
 
 static void accept_connection(h2o_socket_t *listener, const char *err);
 static void accept_http_connection(h2o_socket_t *listener, const char *err);
-static void do_epoll_wait(h2o_socket_t *epoll_sock, const char *err);
 static int get_listener_socket(const char *bind_address, uint16_t port);
 static void on_close_connection(void *data);
 static void process_messages(h2o_multithread_receiver_t *receiver, h2o_linklist_t *messages);
@@ -96,32 +93,6 @@ static void accept_http_connection(h2o_socket_t *listener, const char *err)
 	ctx->event_loop.h2o_accept_ctx.ssl_ctx = ssl_ctx;
 }
 
-static void do_epoll_wait(h2o_socket_t *epoll_sock, const char *err)
-{
-	if (err)
-		ERROR(err);
-	else {
-		thread_context_t * const ctx = H2O_STRUCT_FROM_MEMBER(thread_context_t,
-		                                                      event_loop,
-		                                                      epoll_sock->data);
-		int ready;
-		struct epoll_event event[MAX_EPOLL_EVENTS];
-
-		do
-			ready = epoll_wait(ctx->event_loop.epoll_fd, event, ARRAY_SIZE(event), 0);
-		while (ready < 0 && errno == EINTR);
-
-		if (ready > 0)
-			do {
-				void (** const on_write_ready)(void *) = event[--ready].data.ptr;
-
-				(*on_write_ready)(on_write_ready);
-			} while (ready > 0);
-		else if (ready < 0)
-			STANDARD_ERROR("epoll_wait");
-	}
-}
-
 static int get_listener_socket(const char *bind_address, uint16_t port)
 {
 	int ret = -1;
@@ -260,7 +231,6 @@ void free_event_loop(event_loop_t *event_loop, h2o_multithread_receiver_t *h2o_r
 
 	h2o_multithread_unregister_receiver(event_loop->h2o_ctx.queue, h2o_receiver);
 	h2o_socket_close(event_loop->h2o_socket);
-	h2o_socket_close(event_loop->epoll_socket);
 
 	h2o_loop_t * const loop = event_loop->h2o_ctx.loop;
 
@@ -293,15 +263,6 @@ void initialize_event_loop(bool is_main_thread,
 	h2o_multithread_register_receiver(loop->h2o_ctx.queue,
 	                                  h2o_receiver,
 	                                  process_messages);
-	// libh2o's event loop does not support write polling unless it
-	// controls sending the data as well, so do read polling on the
-	// epoll file descriptor as a work-around.
-	CHECK_ERRNO_RETURN(loop->epoll_fd, epoll_create1, EPOLL_CLOEXEC);
-	loop->epoll_socket = h2o_evloop_socket_create(loop->h2o_ctx.loop,
-	                                              loop->epoll_fd,
-	                                              H2O_SOCKET_FLAG_DONT_READ);
-	loop->epoll_socket->data = loop;
-	h2o_socket_read_start(loop->epoll_socket, do_epoll_wait);
 
 	if (is_main_thread) {
 		global_data->signals = h2o_evloop_socket_create(loop->h2o_ctx.loop,
@@ -311,30 +272,3 @@ void initialize_event_loop(bool is_main_thread,
 		h2o_socket_read_start(global_data->signals, shutdown_server);
 	}
 }
-
-int start_write_polling(int fd,
-                        void (**on_write_ready)(void *),
-                        bool rearm,
-                        event_loop_t *event_loop)
-{
-	struct epoll_event event;
-
-	memset(&event, 0, sizeof(event));
-	event.data.ptr = on_write_ready;
-	event.events = EPOLLET | EPOLLONESHOT | EPOLLOUT | EPOLLRDHUP;
-
-	const int ret = epoll_ctl(event_loop->epoll_fd,
-	                          rearm ? EPOLL_CTL_MOD : EPOLL_CTL_ADD,
-	                          fd,
-	                          &event);
-
-	if (ret)
-		STANDARD_ERROR("epoll_ctl");
-
-	return ret;
-}
-
-void stop_write_polling(int fd, event_loop_t *event_loop)
-{
-	epoll_ctl(event_loop->epoll_fd, EPOLL_CTL_DEL, fd, NULL);
-}

frameworks/C/h2o/src/event_loop.h

@@ -29,11 +29,9 @@
 typedef struct thread_context_t thread_context_t;
 
 typedef struct {
-	h2o_socket_t *epoll_socket;
 	h2o_socket_t *h2o_https_socket;
 	h2o_socket_t *h2o_socket;
 	size_t conn_num;
-	int epoll_fd;
 	h2o_accept_ctx_t h2o_accept_ctx;
 	h2o_context_t h2o_ctx;
 } event_loop_t;
@@ -44,10 +42,5 @@ void initialize_event_loop(bool is_main_thread,
                            global_data_t *global_data,
                            h2o_multithread_receiver_t *h2o_receiver,
                            event_loop_t *loop);
-int start_write_polling(int fd,
-                        void (**on_write_ready)(void *),
-                        bool rearm,
-                        event_loop_t *event_loop);
-void stop_write_polling(int fd, event_loop_t *event_loop);
 
 #endif // EVENT_LOOP_H_

frameworks/C/h2o/src/thread.c

@@ -26,7 +26,6 @@
 #include <stdlib.h>
 #include <string.h>
 #include <h2o/serverutil.h>
-#include <sys/epoll.h>
 #include <sys/syscall.h>
 
 #include "database.h"