H2O: Switch back to a global shared cache (#4869)

frameworks/C/h2o/src/event_loop.c

@@ -63,7 +63,7 @@ static void accept_connection(h2o_socket_t *listener, const char *err)
 		                                                      event_loop,
 		                                                      listener->data);
 
-		if (!ctx->global_data->shutdown) {
+		if (!ctx->shutdown) {
 			size_t accepted = ctx->config->max_accept;
 
 			do {
@@ -170,25 +170,39 @@ static void on_close_connection(void *data)
 
 static void process_messages(h2o_multithread_receiver_t *receiver, h2o_linklist_t *messages)
 {
-	IGNORE_FUNCTION_PARAMETER(messages);
-
 	global_thread_data_t * const global_thread_data = H2O_STRUCT_FROM_MEMBER(global_thread_data_t,
 	                                                                         h2o_receiver,
 	                                                                         receiver);
 
-	// Close the listening sockets immediately, so that if another instance of
-	// the application is started before the current one exits (e.g. when doing
-	// an update), it will accept all incoming connections.
-	if (global_thread_data->ctx->event_loop.h2o_https_socket) {
-		h2o_socket_read_stop(global_thread_data->ctx->event_loop.h2o_https_socket);
-		h2o_socket_close(global_thread_data->ctx->event_loop.h2o_https_socket);
-		global_thread_data->ctx->event_loop.h2o_https_socket = NULL;
-	}
+	while (!h2o_linklist_is_empty(messages)) {
+		message_t * const msg = H2O_STRUCT_FROM_MEMBER(message_t, super, messages->next);
+
+		h2o_linklist_unlink(&msg->super.link);
+
+		switch (msg->type) {
+			case SHUTDOWN:
+				// Close the listening sockets immediately, so that if another instance of
+				// the application is started before the current one exits (e.g. when doing
+				// an update), it will accept all incoming connections.
+				if (global_thread_data->ctx->event_loop.h2o_https_socket) {
+					h2o_socket_read_stop(global_thread_data->ctx->event_loop.h2o_https_socket);
+					h2o_socket_close(global_thread_data->ctx->event_loop.h2o_https_socket);
+					global_thread_data->ctx->event_loop.h2o_https_socket = NULL;
+				}
+
+				if (global_thread_data->ctx->event_loop.h2o_socket) {
+					h2o_socket_read_stop(global_thread_data->ctx->event_loop.h2o_socket);
+					h2o_socket_close(global_thread_data->ctx->event_loop.h2o_socket);
+					global_thread_data->ctx->event_loop.h2o_socket = NULL;
+				}
+
+				global_thread_data->ctx->shutdown = true;
+				break;
+			default:
+				break;
+		}
 
-	if (global_thread_data->ctx->event_loop.h2o_socket) {
-		h2o_socket_read_stop(global_thread_data->ctx->event_loop.h2o_socket);
-		h2o_socket_close(global_thread_data->ctx->event_loop.h2o_socket);
-		global_thread_data->ctx->event_loop.h2o_socket = NULL;
+		free(msg);
 	}
 }
 
@@ -201,7 +215,7 @@ static void shutdown_server(h2o_socket_t *listener, const char *err)
 		                                                      event_loop,
 		                                                      listener->data);
 
-		ctx->global_data->shutdown = true;
+		ctx->shutdown = true;
 
 		// Close the listening sockets immediately, so that if another instance
 		// of the application is started before the current one exits (e.g. when
@@ -218,8 +232,15 @@ static void shutdown_server(h2o_socket_t *listener, const char *err)
 			ctx->event_loop.h2o_socket = NULL;
 		}
 
-		for (size_t i = ctx->config->thread_num - 1; i > 0; i--)
-			h2o_multithread_send_message(&ctx->global_thread_data[i].h2o_receiver, NULL);
+		for (size_t i = ctx->config->thread_num - 1; i > 0; i--) {
+			message_t * const msg = calloc(1, sizeof(*msg));
+
+			if (!msg)
+				abort();
+
+			msg->type = SHUTDOWN;
+			h2o_multithread_send_message(&ctx->global_thread_data[i].h2o_receiver, &msg->super);
+		}
 	}
 }
 
@@ -248,7 +269,7 @@ static void start_accept_polling(const config_t *config,
 
 void event_loop(thread_context_t *ctx)
 {
-	while (!ctx->global_data->shutdown || ctx->event_loop.conn_num)
+	while (!ctx->shutdown || ctx->event_loop.conn_num)
 		h2o_evloop_run(ctx->event_loop.h2o_ctx.loop, INT32_MAX);
 }
 

frameworks/C/h2o/src/event_loop.h

@@ -26,6 +26,10 @@
 
 #include "global_data.h"
 
+typedef enum {
+	SHUTDOWN
+} message_type_t;
+
 typedef struct thread_context_t thread_context_t;
 
 typedef struct {
@@ -36,6 +40,11 @@ typedef struct {
 	h2o_context_t h2o_ctx;
 } event_loop_t;
 
+typedef struct {
+	message_type_t type;
+	h2o_multithread_message_t super;
+} message_t;
+
 void event_loop(thread_context_t *ctx);
 void free_event_loop(event_loop_t *event_loop, h2o_multithread_receiver_t *h2o_receiver);
 void initialize_event_loop(bool is_main_thread,

frameworks/C/h2o/src/global_data.h

@@ -31,7 +31,8 @@
 #include "list.h"
 #include "handlers/request_handler_data.h"
 
-typedef struct global_thread_data_t global_thread_data_t;
+struct global_thread_data_t;
+struct thread_context_t;
 
 typedef struct {
 	const char *bind_address;
@@ -52,15 +53,19 @@ typedef struct {
 
 typedef struct {
 	h2o_logger_t *file_logger;
-	global_thread_data_t *global_thread_data;
+	struct global_thread_data_t *global_thread_data;
+	list_t *postinitialization_tasks;
 	list_t *prepared_statements;
 	h2o_socket_t *signals;
 	SSL_CTX *ssl_ctx;
 	size_t memory_alignment;
 	int signal_fd;
-	bool shutdown;
 	h2o_globalconf_t h2o_config;
 	request_handler_data_t request_handler_data;
 } global_data_t;
 
+void add_postinitialization_task(void (*task)(struct thread_context_t *, void *),
+                                 void *arg,
+                                 list_t **postinitialization_tasks);
+
 #endif // GLOBAL_DATA_H_

frameworks/C/h2o/src/handlers/request_handler_data.h

@@ -21,18 +21,18 @@
 
 #define REQUEST_HANDLER_DATA_H_
 
-#include <h2o.h>
-#include <stdbool.h>
+#include "cache.h"
 
 struct mustache_token_t;
 
 typedef struct {
 	struct mustache_token_t *fortunes_template;
+	cache_t world_cache;
 } request_handler_data_t;
 
 typedef struct {
-	h2o_cache_t *world_cache;
-	bool populate_world_cache;
+	// Replace with any actual fields; structures without members cause compiler warnings.
+	int pad;
 } request_handler_thread_data_t;
 
 #endif // REQUEST_HANDLER_DATA_H_

frameworks/C/h2o/src/handlers/world.c

@@ -34,6 +34,7 @@
 #include "cache.h"
 #include "database.h"
 #include "error.h"
+#include "global_data.h"
 #include "request_handler.h"
 #include "thread.h"
 #include "utility.h"
@@ -120,7 +121,7 @@ static int compare_items(const void *x, const void *y);
 static void complete_multiple_query(multiple_query_ctx_t *query_ctx);
 static int do_multiple_queries(bool do_update, bool use_cache, h2o_req_t *req);
 static void do_updates(multiple_query_ctx_t *query_ctx);
-static void fetch_from_cache(uint64_t now, h2o_cache_t *cache, multiple_query_ctx_t *query_ctx);
+static void fetch_from_cache(uint64_t now, cache_t *cache, multiple_query_ctx_t *query_ctx);
 static void free_cache_entry(h2o_iovec_t value);
 static size_t get_query_number(h2o_req_t *req);
 static void initialize_ids(size_t num_query, query_result_t *res, unsigned int *seed);
@@ -135,7 +136,7 @@ static void on_single_query_error(db_query_param_t *param, const char *error_str
 static result_return_t on_single_query_result(db_query_param_t *param, PGresult *result);
 static void on_single_query_timeout(db_query_param_t *param);
 static result_return_t on_update_result(db_query_param_t *param, PGresult *result);
-static void populate_cache(thread_context_t *ctx);
+static void populate_cache(thread_context_t *ctx, void *arg);
 static void process_result(PGresult *result, size_t idx, query_result_t *out);
 static int serialize_item(uint32_t id, uint32_t random_number, yajl_gen gen);
 static void serialize_items(const query_result_t *res,
@@ -267,12 +268,9 @@ static int do_multiple_queries(bool do_update, bool use_cache, h2o_req_t *req)
 			query_ctx->flags |= DO_UPDATE;
 
 		if (use_cache) {
-			if (ctx->request_handler_data.populate_world_cache)
-				populate_cache(ctx);
-
 			query_ctx->flags |= USE_CACHE;
 			fetch_from_cache(h2o_now(ctx->event_loop.h2o_ctx.loop),
-			                 ctx->request_handler_data.world_cache,
+			                 &ctx->global_data->request_handler_data.world_cache,
 			                 query_ctx);
 
 			if (query_ctx->num_result == query_ctx->num_query) {
@@ -380,21 +378,22 @@ error:
 	send_error(INTERNAL_SERVER_ERROR, REQ_ERROR, query_ctx->req);
 }
 
-static void fetch_from_cache(uint64_t now, h2o_cache_t *cache, multiple_query_ctx_t *query_ctx)
+static void fetch_from_cache(uint64_t now, cache_t *cache, multiple_query_ctx_t *query_ctx)
 {
 	h2o_iovec_t key = {.len = sizeof(query_ctx->res->id)};
 
 	for (size_t i = 0; i < query_ctx->num_query; i++) {
 		key.base = (char *) &query_ctx->res[i].id;
 
-		h2o_cache_ref_t * const r = h2o_cache_fetch(cache, now, key, 0);
+		const h2o_cache_hashcode_t keyhash = h2o_cache_calchash(key.base, key.len);
+		h2o_cache_ref_t * const r = cache_fetch(cache, now, key, keyhash);
 
 		if (r) {
 			query_ctx->res[i].id = query_ctx->res[query_ctx->num_result].id;
 			memcpy(query_ctx->res + query_ctx->num_result++,
 			       r->value.base,
 			       sizeof(*query_ctx->res));
-			h2o_cache_release(cache, r);
+			cache_release(cache, r, keyhash);
 		}
 	}
 }
@@ -490,11 +489,11 @@ static result_return_t on_multiple_query_result(db_query_param_t *param, PGresul
 				const h2o_iovec_t value = {.base = (char *) r, .len = sizeof(*r)};
 
 				*r = query_ctx->res[query_ctx->num_result];
-				h2o_cache_set(query_ctx->ctx->request_handler_data.world_cache,
-				              h2o_now(query_ctx->ctx->event_loop.h2o_ctx.loop),
-				              key,
-				              0,
-				              value);
+				cache_set(h2o_now(query_ctx->ctx->event_loop.h2o_ctx.loop),
+				          key,
+				          0,
+				          value,
+				          &query_ctx->ctx->global_data->request_handler_data.world_cache);
 			}
 		}
 
@@ -568,11 +567,11 @@ static result_return_t on_populate_cache_result(db_query_param_t *param, PGresul
 				const h2o_iovec_t key = {.base = (char *) &r->id, .len = sizeof(r->id)};
 				const h2o_iovec_t value = {.base = (char *) r, .len = sizeof(*r)};
 
-				h2o_cache_set(query_ctx->ctx->request_handler_data.world_cache,
-				              h2o_now(query_ctx->ctx->event_loop.h2o_ctx.loop),
-				              key,
-				              0,
-				              value);
+				cache_set(h2o_now(query_ctx->ctx->event_loop.h2o_ctx.loop),
+				          key,
+				          0,
+				          value,
+				          &query_ctx->ctx->global_data->request_handler_data.world_cache);
 			}
 		}
 	}
@@ -693,8 +692,10 @@ static result_return_t on_update_result(db_query_param_t *param, PGresult *resul
 	return DONE;
 }
 
-static void populate_cache(thread_context_t *ctx)
+static void populate_cache(thread_context_t *ctx, void *arg)
 {
+	IGNORE_FUNCTION_PARAMETER(arg);
+
 	populate_cache_ctx_t * const query_ctx = calloc(1, sizeof(*query_ctx));
 
 	if (query_ctx) {
@@ -706,8 +707,6 @@ static void populate_cache(thread_context_t *ctx)
 
 		if (execute_query(ctx, &query_ctx->param))
 			free(query_ctx);
-		else
-			ctx->request_handler_data.populate_world_cache = false;
 	}
 }
 
@@ -833,30 +832,27 @@ static int updates(struct st_h2o_handler_t *self, h2o_req_t *req)
 	return do_multiple_queries(true, false, req);
 }
 
-void free_world_handler_thread_data(request_handler_thread_data_t *request_handler_thread_data)
+void cleanup_world_handlers(global_data_t *global_data)
 {
-	h2o_cache_destroy(request_handler_thread_data->world_cache);
+	cache_destroy(&global_data->request_handler_data.world_cache);
 }
 
-void initialize_world_handler_thread_data(request_handler_thread_data_t *request_handler_data)
-{
-	request_handler_data->populate_world_cache = true;
-	request_handler_data->world_cache = h2o_cache_create(0,
-	                                                     CACHE_CAPACITY,
-	                                                     CACHE_DURATION,
-	                                                     free_cache_entry);
-
-	if (!request_handler_data->world_cache)
-		abort();
-}
-
-void initialize_world_handlers(global_data_t *global_data,
+void initialize_world_handlers(const config_t *config,
+                               global_data_t *global_data,
                                h2o_hostconf_t *hostconf,
                                h2o_access_log_filehandle_t *log_handle)
 {
 	add_prepared_statement(WORLD_TABLE_NAME, WORLD_QUERY, &global_data->prepared_statements);
-	register_request_handler("/cached-worlds", cached_queries, hostconf, log_handle);
 	register_request_handler("/db", single_query, hostconf, log_handle);
 	register_request_handler("/queries", multiple_queries, hostconf, log_handle);
 	register_request_handler("/updates", updates, hostconf, log_handle);
+
+	if (!cache_create(config->thread_num,
+	                  CACHE_CAPACITY,
+	                  CACHE_DURATION,
+	                  free_cache_entry,
+	                  &global_data->request_handler_data.world_cache)) {
+		add_postinitialization_task(populate_cache, NULL, &global_data->postinitialization_tasks);
+		register_request_handler("/cached-worlds", cached_queries, hostconf, log_handle);
+	}
 }

frameworks/C/h2o/src/handlers/world.h

@@ -25,9 +25,9 @@
 
 #include "global_data.h"
 
-void free_world_handler_thread_data(request_handler_thread_data_t *request_handler_thread_data);
-void initialize_world_handler_thread_data(request_handler_thread_data_t *request_handler_data);
-void initialize_world_handlers(global_data_t *global_data,
+void cleanup_world_handlers(global_data_t *global_data);
+void initialize_world_handlers(const config_t *config,
+                               global_data_t *global_data,
                                h2o_hostconf_t *hostconf,
                                h2o_access_log_filehandle_t *log_handle);
 

frameworks/C/h2o/src/main.c

@@ -46,9 +46,16 @@
 	"[-q <max enqueued database queries per thread>] [-r <root directory>] " \
 	"[-s <HTTPS port>] [-t <thread number>]\n"
 
+typedef struct {
+	list_t l;
+	void *arg;
+	void (*task)(thread_context_t *, void *);
+} task_t;
+
 static void free_global_data(global_data_t *global_data);
 static int initialize_global_data(const config_t *config, global_data_t *global_data);
 static int parse_options(int argc, char *argv[], config_t *config);
+static void run_postinitialization_tasks(list_t **tasks, thread_context_t *ctx);
 static void set_default_options(config_t *config);
 static void setup_process(void);
 
@@ -211,6 +218,18 @@ static int parse_options(int argc, char *argv[], config_t *config)
 	return EXIT_SUCCESS;
 }
 
+static void run_postinitialization_tasks(list_t **tasks, thread_context_t *ctx)
+{
+	if (*tasks)
+		do {
+			task_t * const t = H2O_STRUCT_FROM_MEMBER(task_t, l, *tasks);
+
+			*tasks = (*tasks)->next;
+			t->task(ctx, t->arg);
+			free(t);
+		} while (*tasks);
+}
+
 static void set_default_options(config_t *config)
 {
 	if (!config->max_accept)
@@ -253,6 +272,21 @@ static void setup_process(void)
 	CHECK_ERRNO(setrlimit, RLIMIT_NOFILE, &rlim);
 }
 
+void add_postinitialization_task(void (*task)(struct thread_context_t *, void *),
+                                 void *arg,
+                                 list_t **postinitialization_tasks)
+{
+	task_t * const t = calloc(1, sizeof(*t));
+
+	if (!t)
+		abort();
+
+	t->l.next = *postinitialization_tasks;
+	t->arg = arg;
+	t->task = task;
+	*postinitialization_tasks = &t->l;
+}
+
 int main(int argc, char *argv[])
 {
 	config_t config;
@@ -267,6 +301,7 @@ int main(int argc, char *argv[])
 			setup_process();
 			start_threads(global_data.global_thread_data);
 			initialize_thread_context(global_data.global_thread_data, true, &ctx);
+			run_postinitialization_tasks(&global_data.postinitialization_tasks, &ctx);
 			event_loop(&ctx);
 			// Even though this is global data, we need to close
 			// it before the associated event loop is cleaned up.

frameworks/C/h2o/src/request_handler.c

@@ -61,11 +61,12 @@ static const char *status_code_to_string(http_status_code_t status_code)
 void cleanup_request_handlers(global_data_t *global_data)
 {
 	cleanup_fortunes_handler(global_data);
+	cleanup_world_handlers(global_data);
 }
 
 void free_request_handler_thread_data(request_handler_thread_data_t *request_handler_thread_data)
 {
-	free_world_handler_thread_data(request_handler_thread_data);
+	IGNORE_FUNCTION_PARAMETER(request_handler_thread_data);
 }
 
 const char *get_query_param(const char *query,
@@ -97,7 +98,7 @@ void initialize_request_handler_thread_data(
 		const config_t *config, request_handler_thread_data_t *request_handler_thread_data)
 {
 	IGNORE_FUNCTION_PARAMETER(config);
-	initialize_world_handler_thread_data(request_handler_thread_data);
+	IGNORE_FUNCTION_PARAMETER(request_handler_thread_data);
 }
 
 void initialize_request_handlers(const config_t *config,
@@ -108,7 +109,7 @@ void initialize_request_handlers(const config_t *config,
 	initialize_fortunes_handler(config, global_data, hostconf, log_handle);
 	initialize_json_serializer_handler(hostconf, log_handle);
 	initialize_plaintext_handler(hostconf, log_handle);
-	initialize_world_handlers(global_data, hostconf, log_handle);
+	initialize_world_handlers(config, global_data, hostconf, log_handle);
 }
 
 void register_request_handler(const char *path,

frameworks/C/h2o/src/thread.h

@@ -51,6 +51,7 @@ struct thread_context_t {
 	list_t *json_generator;
 	size_t json_generator_num;
 	unsigned random_seed;
+	bool shutdown;
 	db_state_t db_state;
 	event_loop_t event_loop;
 	request_handler_thread_data_t request_handler_data;