From 51053eca91152ada48da401fae8a18b414270ada Mon Sep 17 00:00:00 2001
From: phk <phk@d4fa192b-c00b-0410-8231-f00ffab90ce4>
Date: Mon, 9 Jun 2008 08:12:31 +0000
Subject: [PATCH] Properly lock n_wrk* statics variables without using more
 lock operations.

Have the worker threads add themselves to the pool, while the hold the pool
lock anyway.

Collect stats per pool, and have the decimator aggregate over all pools
and into the shared memory counters.


git-svn-id: svn+ssh://projects.linpro.no/svn/varnish/trunk@2655 d4fa192b-c00b-0410-8231-f00ffab90ce4
---
 varnish-cache/bin/varnishd/cache_pool.c | 48 ++++++++++++++++---------
 1 file changed, 32 insertions(+), 16 deletions(-)

diff --git a/varnish-cache/bin/varnishd/cache_pool.c b/varnish-cache/bin/varnishd/cache_pool.c
index a3af2279..c0cec43e 100644
--- a/varnish-cache/bin/varnishd/cache_pool.c
+++ b/varnish-cache/bin/varnishd/cache_pool.c
@@ -29,7 +29,6 @@
  * $Id$
  *
  * XXX: automatic thread-pool size adaptation.
- * XXX: unlocked stats variables: consider summing over pools in timer thread.
  */
 
 #include "config.h"
@@ -72,7 +71,8 @@ struct wq {
 	VTAILQ_HEAD(, workreq)	overflow;
 	unsigned		nthr;
 	unsigned		nqueue;
-	uintmax_t		drops;
+	uintmax_t		ndrop;
+	uintmax_t		noverflow;
 };
 
 static struct wq		**wq;
@@ -226,30 +226,33 @@ wrk_thread(void *priv)
 
 	VSL(SLT_WorkThread, 0, "%p start", w);
 
+	LOCK(&qp->mtx);
+	qp->nthr++;
 	while (1) {
 		CHECK_OBJ_NOTNULL(w, WORKER_MAGIC);
 		assert(!isnan(w->used));
 
 		/* Process overflow requests, if any */
-		LOCK(&qp->mtx);
 		w->wrq = VTAILQ_FIRST(&qp->overflow);
 		if (w->wrq != NULL) {
 			VTAILQ_REMOVE(&qp->overflow, w->wrq, list);
 			qp->nqueue--;
-			VSL_stats->n_wrk_queue--;	/* XXX: unlocked */
 		} else {
 			VTAILQ_INSERT_HEAD(&qp->idle, w, list);
 			AZ(pthread_cond_wait(&w->cond, &qp->mtx));
 		}
-		UNLOCK(&qp->mtx);
 		if (w->wrq == NULL)
 			break;
+		UNLOCK(&qp->mtx);
 		AN(w->wrq);
 		wrq = w->wrq;
 		AN(wrq->func);
 		wrq->func(w, wrq->priv);
 		w->wrq = NULL;
+		LOCK(&qp->mtx);
 	}
+	qp->nthr--;
+	UNLOCK(&qp->mtx);
 
 	VSL(SLT_WorkThread, 0, "%p end", w);
 	if (w->vcl != NULL)
@@ -302,15 +305,13 @@ WRK_Queue(struct workreq *wrq)
 	/* If we have too much in the overflow already, refuse */
 
 	if (qp->nqueue > ovfl_max) {
-		qp->drops++;
-		VSL_stats->n_wrk_drop++;	/* XXX: unlocked */
+		qp->ndrop++;
 		UNLOCK(&qp->mtx);
 		return (-1);
 	}
 
 	VTAILQ_INSERT_TAIL(&qp->overflow, wrq, list);
-	VSL_stats->n_wrk_queue++;	/* XXX: unlocked */
-	VSL_stats->n_wrk_overflow++;	/* XXX: unlocked */
+	qp->noverflow++;
 	qp->nqueue++;
 	UNLOCK(&qp->mtx);
 	AZ(pthread_cond_signal(&herder_cond));
@@ -373,11 +374,11 @@ wrk_addpools(const unsigned pools)
 }
 
 /*--------------------------------------------------------------------
- * If a thread is idle or excess, pick it out of the pool...
+ * If a thread is idle or excess, pick it out of the pool.
  */
 
 static void
-wrk_decimate_flock(struct wq *qp, double t_idle)
+wrk_decimate_flock(struct wq *qp, double t_idle, struct varnish_stats *vs)
 {
 	struct worker *w;
 
@@ -390,14 +391,16 @@ wrk_decimate_flock(struct wq *qp, double t_idle)
 		VTAILQ_REMOVE(&qp->idle, w, list);
 	else
 		w = NULL;
+	vs->n_wrk += qp->nthr;
+	vs->n_wrk_queue += qp->nqueue;
+	vs->n_wrk_drop += qp->ndrop;
+	vs->n_wrk_overflow += qp->noverflow;
 	UNLOCK(&qp->mtx);
 
 	/* And give it a kiss on the cheek... */
 	if (w != NULL) {
 		AZ(w->wrq);
 		AZ(pthread_cond_signal(&w->cond));
-		qp->nthr--;
-		VSL_stats->n_wrk--;	/* XXX: unlocked */
 		(void)usleep(params->wthread_purge_delay * 1000);
 	}
 }
@@ -409,6 +412,7 @@ wrk_decimate_flock(struct wq *qp, double t_idle)
  *  Add pools
  *  Scale constants
  *  Get rid of excess threads
+ *  Aggregate stats across pools
  */
 
 static void *
@@ -416,9 +420,13 @@ wrk_herdtimer_thread(void *priv)
 {
 	volatile unsigned u;
 	double t_idle;
+	struct varnish_stats vsm, *vs;
 
 	THR_Name("wrk_herdtimer");
 
+	memset(&vsm, 0, sizeof vsm);
+	vs = &vsm;
+
 	(void)priv;
 	while (1) {
 		/* Add Pools */
@@ -439,9 +447,19 @@ wrk_herdtimer_thread(void *priv)
 
 		ovfl_max = (nthr_max * params->overflow_max) / 100;
 
+		vs->n_wrk = 0;
+		vs->n_wrk_queue = 0;
+		vs->n_wrk_drop = 0;
+		vs->n_wrk_overflow = 0;
+
 		t_idle = TIM_real() - params->wthread_timeout;
 		for (u = 0; u < nwq; u++)
-			wrk_decimate_flock(wq[u], t_idle);
+			wrk_decimate_flock(wq[u], t_idle, vs);
+
+		VSL_stats->n_wrk= vs->n_wrk;
+		VSL_stats->n_wrk_queue = vs->n_wrk_queue;
+		VSL_stats->n_wrk_drop = vs->n_wrk_drop;
+		VSL_stats->n_wrk_overflow = vs->n_wrk_overflow;
 
 		(void)usleep(params->wthread_purge_delay * 1000);
 	}
@@ -470,9 +488,7 @@ wrk_breed_flock(struct wq *qp)
 			VSL_stats->n_wrk_failed++;
 			(void)usleep(params->wthread_fail_delay * 1000);
 		} else {
-			qp->nthr++;
 			AZ(pthread_detach(tp));
-			VSL_stats->n_wrk++;	/* XXX: unlocked */
 			VSL_stats->n_wrk_create++;
 			(void)usleep(params->wthread_add_delay * 1000);
 		}
-- 
2.39.5