[PATCH] Set the waiter pipe as non-blocking and record overflows

[Tollef Fog Heen]

Fixes #1285
---

I wonder if it would make sense to pull the "set this fd as non-blocking" into a libvarnish function.  Tested (make check) on Linux with epoll and poll.  Not tested on FreeBSD and Solaris.

 bin/varnishd/waiter/cache_waiter_epoll.c  |   19 ++++++++++++++++---
 bin/varnishd/waiter/cache_waiter_kqueue.c |   15 +++++++++++++--
 bin/varnishd/waiter/cache_waiter_poll.c   |   19 +++++++++++++++++--
 bin/varnishd/waiter/cache_waiter_ports.c  |   10 +++++++---
 bin/varnishd/waiter/waiter.h              |    2 +-
 include/tbl/vsc_f_main.h                  |    5 +++++
 6 files changed, 59 insertions(+), 11 deletions(-)

diff --git a/bin/varnishd/waiter/cache_waiter_epoll.c b/bin/varnishd/waiter/cache_waiter_epoll.c
index a15bbe5..35b0e20 100644
--- a/bin/varnishd/waiter/cache_waiter_epoll.c
+++ b/bin/varnishd/waiter/cache_waiter_epoll.c
@@ -39,6 +39,7 @@
 
 #include <fcntl.h>
 #include <stdlib.h>
+#include <math.h>
 
 #include "cache/cache.h"
 
@@ -49,7 +50,7 @@
 #  define EPOLLRDHUP 0
 #endif
 
-#define NEEV	100
+#define NEEV	8192
 
 struct vwe {
 	unsigned		magic;
@@ -220,12 +221,18 @@ vwe_timeout_idle_ticker(void *priv)
 /*--------------------------------------------------------------------*/
 
 static void
-vwe_pass(void *priv, const struct sess *sp)
+vwe_pass(void *priv, struct sess *sp)
 {
 	struct vwe *vwe;
+	int written;
 
 	CAST_OBJ_NOTNULL(vwe, priv, VWE_MAGIC);
-	assert(sizeof sp == write(vwe->pipes[1], &sp, sizeof sp));
+
+	written = write(vwe->pipes[1], &sp, sizeof sp);
+	if (written != sizeof sp) {
+		VSC_C_main->sess_pipe_overflow++;
+		SES_Delete(sp, SC_OVERLOAD, NAN);
+	}
 }
 
 /*--------------------------------------------------------------------*/
@@ -248,6 +255,12 @@ vwe_init(void)
 	i = fcntl(vwe->pipes[0], F_SETFL, i);
 	assert(i != -1);
 
+	i = fcntl(vwe->pipes[1], F_GETFL);
+	assert(i != -1);
+	i |= O_NONBLOCK;
+	i = fcntl(vwe->pipes[1], F_SETFL, i);
+	assert(i != -1);
+
 	i = fcntl(vwe->timer_pipes[0], F_GETFL);
 	assert(i != -1);
 	i |= O_NONBLOCK;
diff --git a/bin/varnishd/waiter/cache_waiter_kqueue.c b/bin/varnishd/waiter/cache_waiter_kqueue.c
index 8d37ad6..20e03a9 100644
--- a/bin/varnishd/waiter/cache_waiter_kqueue.c
+++ b/bin/varnishd/waiter/cache_waiter_kqueue.c
@@ -210,12 +210,17 @@ vwk_thread(void *priv)
 /*--------------------------------------------------------------------*/
 
 static void
-vwk_pass(void *priv, const struct sess *sp)
+vwk_pass(void *priv, struct sess *sp)
 {
 	struct vwk *vwk;
 
 	CAST_OBJ_NOTNULL(vwk, priv, VWK_MAGIC);
-	assert(sizeof sp == write(vwk->pipes[1], &sp, sizeof sp));
+
+	written = write(vwk->pipes[1], &sp, sizeof sp);
+	if (written != sizeof sp) {
+		VSC_C_main->sess_pipe_overflow++;
+		SES_Delete(sp, SC_OVERLOAD, NAN);
+	}
 }
 
 /*--------------------------------------------------------------------*/
@@ -238,6 +243,12 @@ vwk_init(void)
 	i = fcntl(vwk->pipes[0], F_SETFL, i);
 	assert(i != -1);
 
+	i = fcntl(vwe->pipes[1], F_GETFL);
+	assert(i != -1);
+	i |= O_NONBLOCK;
+	i = fcntl(vwe->pipes[1], F_SETFL, i);
+	assert(i != -1);
+
 	AZ(pthread_create(&vwk->thread, NULL, vwk_thread, vwk));
 	return (vwk);
 }
diff --git a/bin/varnishd/waiter/cache_waiter_poll.c b/bin/varnishd/waiter/cache_waiter_poll.c
index 4339617..34bafc2 100644
--- a/bin/varnishd/waiter/cache_waiter_poll.c
+++ b/bin/varnishd/waiter/cache_waiter_poll.c
@@ -30,8 +30,10 @@
 
 #include "config.h"
 
+#include <fcntl.h>
 #include <poll.h>
 #include <stdlib.h>
+#include <math.h>
 
 #include "cache/cache.h"
 
@@ -191,13 +193,18 @@ vwp_main(void *priv)
 /*--------------------------------------------------------------------*/
 
 static void
-vwp_poll_pass(void *priv, const struct sess *sp)
+vwp_poll_pass(void *priv, struct sess *sp)
 {
 	struct vwp *vwp;
+	int written;
 
 	CAST_OBJ_NOTNULL(vwp, priv, VWP_MAGIC);
 
-	assert(sizeof sp == write(vwp->pipes[1], &sp, sizeof sp));
+	written = write(vwp->pipes[1], &sp, sizeof sp);
+	if (written != sizeof sp) {
+		VSC_C_main->sess_pipe_overflow++;
+		SES_Delete(sp, SC_OVERLOAD, NAN);
+	}
 }
 
 /*--------------------------------------------------------------------*/
@@ -205,12 +212,20 @@ vwp_poll_pass(void *priv, const struct sess *sp)
 static void *
 vwp_poll_init(void)
 {
+	int i;
 	struct vwp *vwp;
 
 	ALLOC_OBJ(vwp, VWP_MAGIC);
 	AN(vwp);
 	VTAILQ_INIT(&vwp->sesshead);
 	AZ(pipe(vwp->pipes));
+
+	i = fcntl(vwp->pipes[1], F_GETFL);
+	assert(i != -1);
+	i |= O_NONBLOCK;
+	i = fcntl(vwp->pipes[1], F_SETFL, i);
+	assert(i != -1);
+
 	vwp_pollspace(vwp, 256);
 	AZ(pthread_create(&vwp->poll_thread, NULL, vwp_main, vwp));
 	return (vwp);
diff --git a/bin/varnishd/waiter/cache_waiter_ports.c b/bin/varnishd/waiter/cache_waiter_ports.c
index 3c07b95..e6825f3 100644
--- a/bin/varnishd/waiter/cache_waiter_ports.c
+++ b/bin/varnishd/waiter/cache_waiter_ports.c
@@ -240,14 +240,18 @@ vws_thread(void *priv)
 /*--------------------------------------------------------------------*/
 
 static void
-vws_pass(void *priv, const struct sess *sp)
+vws_pass(void *priv, struct sess *sp)
 {
 	int r;
 	struct vws *vws;
 
 	CAST_OBJ_NOTNULL(vws, priv, VWS_MAGIC);
-	while((r = port_send(vws->dport, 0, TRUST_ME(sp))) == -1 &&
-		errno == EAGAIN);
+	r = port_send(vws->dport, 0, TRUST_ME(sp));
+	if (r == -1 && errno == EAGAIN) {
+		VSC_C_main->sess_pipe_overflow++;
+		SES_Delete(sp, SC_OVERLOAD, NAN);
+		return;
+	}
 	AZ(r);
 }
 
diff --git a/bin/varnishd/waiter/waiter.h b/bin/varnishd/waiter/waiter.h
index 9f9f795..ce46e4d 100644
--- a/bin/varnishd/waiter/waiter.h
+++ b/bin/varnishd/waiter/waiter.h
@@ -31,7 +31,7 @@
 struct sess;
 
 typedef void* waiter_init_f(void);
-typedef void waiter_pass_f(void *priv, const struct sess *);
+typedef void waiter_pass_f(void *priv, struct sess *);
 
 #define WAITER_DEFAULT		"platform dependent"
 
diff --git a/include/tbl/vsc_f_main.h b/include/tbl/vsc_f_main.h
index 3f6d3fe..6506668 100644
--- a/include/tbl/vsc_f_main.h
+++ b/include/tbl/vsc_f_main.h
@@ -88,6 +88,11 @@ VSC_F(sess_fail,		uint64_t, 1, 'c',
 	" some resource like filedescriptors."
 )
 
+VSC_F(sess_pipe_overflow,		uint64_t, 1, 'c',
+    "Session pipe overflow",
+	"Count of sessions dropped due to the session pipe overflowing."
+)
+
 /*---------------------------------------------------------------------*/
 
 VSC_F(client_req_400,		uint64_t, 1, 'a',
-- 
1.7.10.4