Improve very heavy load handling.

Introduces two new configuration knobs:
	* socket_backlog (backlog for listen(2))
	* http_request_limit

The second one is the most interesting one.

Before, kore would iterate over all received HTTP requests
in its queue before returning out of http_process().

Under heavy load this queue can cause Kore to spend a considerable
amount of time iterating over said queue. With the http_request_limit,
kore will process at MOST http_request_limit requests before returning
back to the event loop.

This means responses to processed requests are sent out much quicker
and allows kore to handle any other incoming requests more gracefully.

conf/kore.conf.example

@@ -1,5 +1,9 @@
 # Example Kore configuration
 
+# Maximum length to queue pending connections (see listen(2))
+# MUST be set before any bind directive.
+#socket_backlog			5000
+
 # Server configuration.
 bind		127.0.0.1 443
 
@@ -35,10 +39,14 @@ workers		4
 #	http_hsts_enable	Send Strict Transport Security header in
 #				all responses. Parameter is the age.
 #				(Set to 0 to disable sending this header).
+#
+#	http_request_limit	Limit the number of requests Kore processes
+#				in a single event loop.
 #http_header_max	4096
 #http_body_max		10240000
 #http_keepalive_time	0
 #http_hsts_enable	31536000
+#http_request_limit	1000
 
 # Websocket specific settings.
 #	websocket_maxframe	Specifies the maximum frame size we can receive

includes/http.h

@@ -30,6 +30,7 @@ extern "C" {
 #define HTTP_REQ_HEADER_MAX	25
 #define HTTP_MAX_QUERY_ARGS	10
 #define HTTP_MAX_COOKIES	10
+#define HTTP_REQUEST_LIMIT	1000
 
 #define HTTP_ARG_TYPE_RAW	0
 #define HTTP_ARG_TYPE_BYTE	1
@@ -209,6 +210,7 @@ extern u_int16_t	http_header_max;
 extern u_int64_t	http_body_max;
 extern u_int64_t	http_hsts_enable;
 extern u_int16_t	http_keepalive_time;
+extern u_int32_t	http_request_limit;
 
 void		http_init(void);
 void		http_process(void);

includes/kore.h

@@ -275,7 +275,6 @@ struct kore_worker {
 	u_int8_t			cpu;
 	pid_t				pid;
 	u_int8_t			has_lock;
-	u_int32_t			accept_treshold;
 	struct kore_module_handle	*active_hdlr;
 };
 
@@ -364,6 +363,7 @@ extern u_int32_t		worker_max_connections;
 extern u_int32_t		worker_active_connections;
 extern u_int64_t		kore_websocket_maxframe;
 extern u_int64_t		kore_websocket_timeout;
+extern u_int32_t		kore_socket_backlog;
 
 extern struct listener_head	listeners;
 extern struct kore_worker	*worker;

src/bsd.c

@@ -151,8 +151,7 @@ kore_platform_event_wait(u_int64_t timer)
 		case KORE_TYPE_LISTENER:
 			l = (struct listener *)events[i].udata;
 
-			while (r < worker->accept_treshold &&
-			    worker_active_connections <
+			while (worker_active_connections <
 			    worker_max_connections) {
 				kore_connection_accept(l, &c);
 				if (c == NULL)

src/config.c

@@ -52,6 +52,7 @@ static int		configure_http_header_max(char **);
 static int		configure_http_body_max(char **);
 static int		configure_http_hsts_enable(char **);
 static int		configure_http_keepalive_time(char **);
+static int		configure_http_request_limit(char **);
 static int		configure_validator(char **);
 static int		configure_params(char **);
 static int		configure_validate(char **);
@@ -63,6 +64,7 @@ static int		configure_authentication_value(char **);
 static int		configure_authentication_validator(char **);
 static int		configure_websocket_maxframe(char **);
 static int		configure_websocket_timeout(char **);
+static int		configure_socket_backlog(char **);
 
 #if defined(KORE_USE_PGSQL)
 static int		configure_pgsql_conn_max(char **);
@@ -99,6 +101,7 @@ static struct {
 	{ "http_body_max",		configure_http_body_max },
 	{ "http_hsts_enable",		configure_http_hsts_enable },
 	{ "http_keepalive_time",	configure_http_keepalive_time },
+	{ "http_request_limit",		configure_http_request_limit },
 	{ "validator",			configure_validator },
 	{ "params",			configure_params },
 	{ "validate",			configure_validate },
@@ -109,6 +112,7 @@ static struct {
 	{ "authentication_validator",	configure_authentication_validator },
 	{ "websocket_maxframe",		configure_websocket_maxframe },
 	{ "websocket_timeout",		configure_websocket_timeout },
+	{ "socket_backlog",		configure_socket_backlog },
 #if defined(KORE_USE_PGSQL)
 	{ "pgsql_conn_max",		configure_pgsql_conn_max },
 #endif
@@ -656,6 +660,23 @@ configure_http_keepalive_time(char **argv)
 	return (KORE_RESULT_OK);
 }
 
+static int
+configure_http_request_limit(char **argv)
+{
+	int		err;
+
+	if (argv[1] == NULL)
+		return (KORE_RESULT_ERROR);
+
+	http_request_limit = kore_strtonum(argv[1], 10, 0, UINT_MAX, &err);
+	if (err != KORE_RESULT_OK) {
+		printf("bad http_request_limit value: %s\n", argv[1]);
+		return (KORE_RESULT_ERROR);
+	}
+
+	return (KORE_RESULT_OK);
+}
+
 static int
 configure_validator(char **argv)
 {
@@ -927,6 +948,23 @@ configure_websocket_timeout(char **argv)
 	return (KORE_RESULT_OK);
 }
 
+static int
+configure_socket_backlog(char **argv)
+{
+	int		err;
+
+	if (argv[1] == NULL)
+		return (KORE_RESULT_ERROR);
+
+	kore_socket_backlog = kore_strtonum(argv[1], 10, 0, UINT_MAX, &err);
+	if (err != KORE_RESULT_OK) {
+		printf("bad socket_backlog value: %s\n", argv[1]);
+		return (KORE_RESULT_ERROR);
+	}
+
+	return (KORE_RESULT_OK);
+}
+
 static void
 domain_sslstart(void)
 {

src/http.c

@@ -53,7 +53,8 @@ static TAILQ_HEAD(, http_request)	http_requests_sleeping;
 static struct kore_pool			http_request_pool;
 static struct kore_pool			http_header_pool;
 
-int		http_request_count;
+int		http_request_count = 0;
+u_int32_t	http_request_limit = HTTP_REQUEST_LIMIT;
 u_int64_t	http_hsts_enable = HTTP_HSTS_ENABLE;
 u_int16_t	http_header_max = HTTP_HEADER_MAX_LEN;
 u_int16_t	http_keepalive_time = HTTP_KEEPALIVE_TIME;
@@ -64,7 +65,6 @@ http_init(void)
 {
 	int		prealloc, l;
 
-	http_request_count = 0;
 	TAILQ_INIT(&http_requests);
 	TAILQ_INIT(&http_requests_sleeping);
 
@@ -219,11 +219,15 @@ http_request_wakeup(struct http_request *req)
 void
 http_process(void)
 {
+	u_int32_t			count;
 	struct http_request		*req, *next;
 
+	count = 0;
 	for (req = TAILQ_FIRST(&http_requests); req != NULL; req = next) {
-		next = TAILQ_NEXT(req, list);
+		if (count >= http_request_limit)
+			break;
 
+		next = TAILQ_NEXT(req, list);
 		if (req->flags & HTTP_REQUEST_DELETE) {
 			http_request_free(req);
 			continue;
@@ -236,6 +240,7 @@ http_process(void)
 		if (!(req->flags & HTTP_REQUEST_COMPLETE))
 			continue;
 
+		count++;
 		http_process_request(req, 0);
 	}
 }

src/kore.c

@@ -34,6 +34,7 @@ int			skip_chroot = 0;
 u_int8_t		worker_count = 0;
 char			*runas_user = NULL;
 char			*chroot_path = NULL;
+u_int32_t		kore_socket_backlog = 5000;
 char			*kore_pidfile = KORE_PIDFILE_DEFAULT;
 char			*kore_ssl_cipher_list = KORE_DEFAULT_CIPHER_LIST;
 
@@ -258,7 +259,7 @@ kore_server_bind(const char *ip, const char *port)
 
 	freeaddrinfo(results);
 
-	if (listen(l->fd, 5000) == -1) {
+	if (listen(l->fd, kore_socket_backlog) == -1) {
 		close(l->fd);
 		kore_mem_free(l);
 		kore_debug("listen(): %s", errno_s);

src/linux.c

@@ -128,8 +128,7 @@ kore_platform_event_wait(u_int64_t timer)
 		case KORE_TYPE_LISTENER:
 			l = (struct listener *)events[i].data.ptr;
 
-			while (r < worker->accept_treshold &&
-			    worker_active_connections <
+			while (worker_active_connections <
 			    worker_max_connections) {
 				kore_connection_accept(l, &c);
 				if (c == NULL)

src/worker.c

@@ -258,11 +258,6 @@ kore_worker_entry(struct kore_worker *kw)
 	kore_log(LOG_NOTICE, "worker %d started (cpu#%d)", kw->id, kw->cpu);
 	kore_module_onload();
 
-	if (worker_count > 1)
-		kw->accept_treshold = worker_max_connections / 16;
-	else
-		kw->accept_treshold = worker_max_connections;
-
 	for (;;) {
 		if (sig_recv != 0) {
 			if (sig_recv == SIGHUP)