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

diff --git a/varnish-cache/bin/varnishd/cache_pool.c b/varnish-cache/bin/varnishd/cache_pool.c
index a3af22793b43b102b81b0dc9f45cd12ff57ea86e..c0cec43e9ac8742738cce8f4607771d64d4969e2 100644 (file)
--- 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);
                }