[experimental-ims] ac393c4 Revamp the code which rides herd on the thread pools:
Geoff Simmons
geoff at varnish-cache.org
Mon Jan 9 21:51:56 CET 2012
commit ac393c4b825a23abc428b4226920a0adc8bec67e
Author: Poul-Henning Kamp <phk at FreeBSD.org>
Date: Sun Sep 18 22:48:01 2011 +0000
Revamp the code which rides herd on the thread pools:
Make one single thread, which creates thread pools, and since that
is not particularly ardous work, also have it maintain the stats
gauge of number of sessions queued.
(Removing pools should now be possible, but the code is not there yet.)
Each pool has its own herder thread, which creates and destroys
threads as required, and updates stats for dropped and queued
sessions.
Sanitize VSC fields associated with this area.
diff --git a/bin/varnishd/cache.h b/bin/varnishd/cache.h
index 9de746c..4f18a08 100644
--- a/bin/varnishd/cache.h
+++ b/bin/varnishd/cache.h
@@ -821,7 +821,7 @@ void Lck__Assert(const struct lock *lck, int held);
/* public interface: */
void LCK_Init(void);
void Lck_Delete(struct lock *lck);
-void Lck_CondWait(pthread_cond_t *cond, struct lock *lck);
+int Lck_CondWait(pthread_cond_t *cond, struct lock *lck, struct timespec *ts);
#define Lck_New(a, b) Lck__New(a, b, #b)
#define Lck_Lock(a) Lck__Lock(a, __func__, __FILE__, __LINE__)
diff --git a/bin/varnishd/cache_lck.c b/bin/varnishd/cache_lck.c
index 3e0c640..b9229b3 100644
--- a/bin/varnishd/cache_lck.c
+++ b/bin/varnishd/cache_lck.c
@@ -139,19 +139,26 @@ Lck__Assert(const struct lock *lck, int held)
!pthread_equal(ilck->owner, pthread_self()));
}
-void __match_proto__()
-Lck_CondWait(pthread_cond_t *cond, struct lock *lck)
+int __match_proto__()
+Lck_CondWait(pthread_cond_t *cond, struct lock *lck, struct timespec *ts)
{
struct ilck *ilck;
+ int retval = 0;
CAST_OBJ_NOTNULL(ilck, lck->priv, ILCK_MAGIC);
AN(ilck->held);
assert(pthread_equal(ilck->owner, pthread_self()));
ilck->held = 0;
- AZ(pthread_cond_wait(cond, &ilck->mtx));
+ if (ts == NULL) {
+ AZ(pthread_cond_wait(cond, &ilck->mtx));
+ } else {
+ retval = pthread_cond_timedwait(cond, &ilck->mtx, ts);
+ assert(retval == 0 || retval == ETIMEDOUT);
+ }
AZ(ilck->held);
ilck->held = 1;
ilck->owner = pthread_self();
+ return (retval);
}
void
diff --git a/bin/varnishd/cache_pool.c b/bin/varnishd/cache_pool.c
index b593617..2aeb1b9 100644
--- a/bin/varnishd/cache_pool.c
+++ b/bin/varnishd/cache_pool.c
@@ -75,6 +75,11 @@ struct pool {
unsigned magic;
#define POOL_MAGIC 0x606658fa
VTAILQ_ENTRY(pool) list;
+
+ pthread_cond_t herder_cond;
+ struct lock herder_mtx;
+ pthread_t herder_thr;
+
struct lock mtx;
struct workerhead idle;
VTAILQ_HEAD(, sess) queue;
@@ -82,20 +87,13 @@ struct pool {
unsigned nthr;
unsigned lqueue;
unsigned last_lqueue;
- uintmax_t ndrop;
- uintmax_t nqueue;
+ uintmax_t ndropped;
+ uintmax_t nqueued;
struct sesspool *sesspool;
};
-static VTAILQ_HEAD(,pool) pools = VTAILQ_HEAD_INITIALIZER(pools);
-
-static unsigned queue_max;
-static unsigned nthr_max;
-
-static unsigned nwq;
-
-static pthread_cond_t herder_cond;
-static struct lock herder_mtx;
+static struct lock pool_mtx;
+static pthread_t thr_pool_herder;
/*--------------------------------------------------------------------
* Nobody is accepting on this socket, so we do.
@@ -163,7 +161,6 @@ Pool_Work_Thread(void *priv, struct worker *w)
CAST_OBJ_NOTNULL(pp, priv, POOL_MAGIC);
w->pool = pp;
Lck_Lock(&pp->mtx);
- pp->nthr++;
stats_clean = 1;
while (1) {
@@ -196,7 +193,7 @@ Pool_Work_Thread(void *priv, struct worker *w)
VTAILQ_INSERT_HEAD(&pp->idle, w, list);
if (!stats_clean)
WRK_SumStat(w);
- Lck_CondWait(&w->cond, &pp->mtx);
+ (void)Lck_CondWait(&w->cond, &pp->mtx, NULL);
}
/*
@@ -251,8 +248,6 @@ Pool_Work_Thread(void *priv, struct worker *w)
stats_clean = WRK_TrySumStat(w);
Lck_Lock(&pp->mtx);
}
- assert(pp->nthr > 0);
- pp->nthr--;
Lck_Unlock(&pp->mtx);
w->pool = NULL;
}
@@ -281,17 +276,17 @@ WRK_Queue(struct pool *pp, struct sess *sp)
}
/* If we have too much in the queue already, refuse. */
- if (pp->lqueue > queue_max) {
- pp->ndrop++;
+ if (pp->lqueue > (params->queue_max * pp->nthr) / 100) {
+ pp->ndropped++;
Lck_Unlock(&pp->mtx);
return (-1);
}
VTAILQ_INSERT_TAIL(&pp->queue, sp, poollist);
- pp->nqueue++;
+ pp->nqueued++;
pp->lqueue++;
Lck_Unlock(&pp->mtx);
- AZ(pthread_cond_signal(&herder_cond));
+ AZ(pthread_cond_signal(&pp->herder_cond));
return (0);
}
@@ -325,149 +320,24 @@ Pool_Schedule(struct pool *pp, struct sess *sp)
}
/*--------------------------------------------------------------------
- * Add (more) thread pools
- */
-
-static struct pool *
-pool_mkpool(void)
-{
- struct pool *pp;
- struct listen_sock *ls;
- struct poolsock *ps;
-
- ALLOC_OBJ(pp, POOL_MAGIC);
- XXXAN(pp);
- Lck_New(&pp->mtx, lck_wq);
- VTAILQ_INIT(&pp->queue);
- VTAILQ_INIT(&pp->idle);
- VTAILQ_INIT(&pp->socks);
- pp->sesspool = SES_NewPool(pp);
- AN(pp->sesspool);
-
- VTAILQ_FOREACH(ls, &heritage.socks, list) {
- if (ls->sock < 0)
- continue;
- ALLOC_OBJ(ps, POOLSOCK_MAGIC);
- XXXAN(ps);
- ps->lsock = ls;
- VTAILQ_INSERT_TAIL(&pp->socks, ps, list);
- }
- VTAILQ_INSERT_TAIL(&pools, pp, list);
- return (pp);
-}
-
-static void
-wrk_addpools(const unsigned npools)
-{
- struct pool *pp;
- unsigned u;
-
- for (u = nwq; u < npools; u++) {
- pp = pool_mkpool();
- XXXAN(pp);
- }
- nwq = npools;
-}
-
-/*--------------------------------------------------------------------
- * If a thread is idle or excess, pick it out of the pool.
- */
-
-static void
-wrk_decimate_flock(struct pool *qp, double t_idle, struct VSC_C_main *vs)
-{
- struct worker *w = NULL;
-
- Lck_Lock(&qp->mtx);
- vs->n_wrk += qp->nthr;
- vs->n_wrk_lqueue += qp->lqueue;
- vs->n_wrk_drop += qp->ndrop;
- vs->n_wrk_queued += qp->nqueue;
-
- if (qp->nthr > params->wthread_min) {
- w = VTAILQ_LAST(&qp->idle, workerhead);
- if (w != NULL && (w->lastused < t_idle || qp->nthr > nthr_max))
- VTAILQ_REMOVE(&qp->idle, w, list);
- else
- w = NULL;
- }
- Lck_Unlock(&qp->mtx);
-
- /* And give it a kiss on the cheek... */
- if (w != NULL) {
- AZ(w->sp);
- AZ(pthread_cond_signal(&w->cond));
- TIM_sleep(params->wthread_purge_delay * 1e-3);
- }
-}
-
-/*--------------------------------------------------------------------
- * Periodic pool herding thread
- *
- * Do things which we can do at our leisure:
- * Add pools
- * Scale constants
- * Get rid of excess threads
- * Aggregate stats across pools
+ * Wait for another request
*/
-static void *
-wrk_herdtimer_thread(void *priv)
+void
+Pool_Wait(struct sess *sp)
{
- volatile unsigned u;
- double t_idle;
- struct VSC_C_main vsm, *vs;
- int errno_is_multi_threaded;
- struct pool *pp;
-
- THR_SetName("wrk_herdtimer");
+ CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
+ AZ(sp->obj);
+ AZ(sp->vcl);
+ assert(sp->fd >= 0);
/*
- * This is one of the first threads created, test to see that
- * errno is really per thread. If this fails, your C-compiler
- * needs some magic argument (-mt, -pthread, -pthreads etc etc).
+ * Set nonblocking in the worker-thread, before passing to the
+ * acceptor thread, to reduce syscall density of the latter.
*/
- errno = 0;
- AN(unlink("/")); /* This had better fail */
- errno_is_multi_threaded = errno;
- assert(errno_is_multi_threaded != 0);
-
- memset(&vsm, 0, sizeof vsm);
- vs = &vsm;
-
- (void)priv;
- while (1) {
- /* Add Pools */
- u = params->wthread_pools;
- if (u > nwq)
- wrk_addpools(u);
-
- /* Scale parameters */
-
- u = params->wthread_max;
- if (u < params->wthread_min)
- u = params->wthread_min;
- nthr_max = u;
-
- queue_max = (nthr_max * params->queue_max) / 100;
-
- vs->n_wrk = 0;
- vs->n_wrk_lqueue = 0;
- vs->n_wrk_drop = 0;
- vs->n_wrk_queued = 0;
-
- t_idle = TIM_real() - params->wthread_timeout;
- VTAILQ_FOREACH(pp, &pools, list)
- wrk_decimate_flock(pp, t_idle, vs);
-
- VSC_C_main->n_wrk= vs->n_wrk;
- VSC_C_main->n_wrk_lqueue = vs->n_wrk_lqueue;
- VSC_C_main->n_wrk_drop = vs->n_wrk_drop;
- VSC_C_main->n_wrk_queued = vs->n_wrk_queued;
-
- TIM_sleep(params->wthread_purge_delay * 1e-3);
- }
- NEEDLESS_RETURN(NULL);
+ if (VTCP_nonblocking(sp->fd))
+ SES_Close(sp, "remote closed");
+ waiter->pass(waiter_priv, sp);
}
/*--------------------------------------------------------------------
@@ -486,23 +356,33 @@ wrk_breed_flock(struct pool *qp, const pthread_attr_t *tp_attr)
if (qp->nthr < params->wthread_min || /* Not enough threads yet */
(qp->lqueue > params->wthread_add_threshold && /* more needed */
qp->lqueue > qp->last_lqueue)) { /* not getting better since last */
- if (qp->nthr >= nthr_max) {
- VSC_C_main->n_wrk_max++;
+ if (qp->nthr > params->wthread_max) {
+ Lck_Lock(&pool_mtx);
+ VSC_C_main->threads_limited++;
+ Lck_Unlock(&pool_mtx);
} else if (pthread_create(&tp, tp_attr, WRK_thread, qp)) {
VSL(SLT_Debug, 0, "Create worker thread failed %d %s",
errno, strerror(errno));
- VSC_C_main->n_wrk_failed++;
+ Lck_Lock(&pool_mtx);
+ VSC_C_main->threads_limited++;
+ Lck_Unlock(&pool_mtx);
TIM_sleep(params->wthread_fail_delay * 1e-3);
} else {
AZ(pthread_detach(tp));
- VSC_C_main->n_wrk_create++;
TIM_sleep(params->wthread_add_delay * 1e-3);
+ qp->nthr++;
+ Lck_Lock(&pool_mtx);
+ VSC_C_main->threads++;
+ VSC_C_main->threads_created++;
+ Lck_Unlock(&pool_mtx);
}
}
qp->last_lqueue = qp->lqueue;
}
/*--------------------------------------------------------------------
+ * Herd a single pool
+ *
* This thread wakes up whenever a pool queues.
*
* The trick here is to not be too aggressive about creating threads.
@@ -515,62 +395,154 @@ wrk_breed_flock(struct pool *qp, const pthread_attr_t *tp_attr)
*
*/
-static void *
-wrk_herder_thread(void *priv)
+static void*
+pool_herder(void *priv)
{
+ struct pool *pp;
pthread_attr_t tp_attr;
- struct pool *pp, *pp2;
+ struct timespec ts;
+ double t_idle;
+ struct worker *w;
+ int i;
- /* Set the stacksize for worker threads */
+ CAST_OBJ_NOTNULL(pp, priv, POOL_MAGIC);
AZ(pthread_attr_init(&tp_attr));
- THR_SetName("wrk_herder");
- (void)priv;
while (1) {
- VTAILQ_FOREACH(pp, &pools, list) {
- if (params->wthread_stacksize != UINT_MAX)
- AZ(pthread_attr_setstacksize(&tp_attr,
- params->wthread_stacksize));
-
- wrk_breed_flock(pp, &tp_attr);
-
- /*
- * Make sure all pools have their minimum complement
- */
- VTAILQ_FOREACH(pp2, &pools, list)
- while (pp2->nthr < params->wthread_min)
- wrk_breed_flock(pp2, &tp_attr);
- /*
- * We cannot avoid getting a mutex, so we have a
- * bogo mutex just for POSIX_STUPIDITY
- */
- Lck_Lock(&herder_mtx);
- Lck_CondWait(&herder_cond, &herder_mtx);
- Lck_Unlock(&herder_mtx);
+ /* Set the stacksize for worker threads we create */
+ if (params->wthread_stacksize != UINT_MAX)
+ AZ(pthread_attr_setstacksize(&tp_attr,
+ params->wthread_stacksize));
+ else {
+ AZ(pthread_attr_destroy(&tp_attr));
+ AZ(pthread_attr_init(&tp_attr));
+ }
+
+ wrk_breed_flock(pp, &tp_attr);
+
+ if (pp->nthr < params->wthread_min)
+ continue;
+
+ AZ(clock_gettime(CLOCK_MONOTONIC, &ts));
+ ts.tv_sec += params->wthread_purge_delay / 1000;
+ ts.tv_nsec +=
+ (params->wthread_purge_delay % 1000) * 1000000;
+ if (ts.tv_nsec >= 1000000000) {
+ ts.tv_sec++;
+ ts.tv_nsec -= 1000000000;
+ }
+
+ Lck_Lock(&pp->herder_mtx);
+ i = Lck_CondWait(&pp->herder_cond, &pp->herder_mtx, &ts);
+ Lck_Unlock(&pp->herder_mtx);
+ if (!i)
+ continue;
+
+ if (pp->nthr <= params->wthread_min)
+ continue;
+
+ t_idle = TIM_real() - params->wthread_timeout;
+
+ Lck_Lock(&pp->mtx);
+ VSC_C_main->sess_queued += pp->nqueued;
+ VSC_C_main->sess_dropped += pp->ndropped;
+ pp->nqueued = pp->ndropped = 0;
+ w = VTAILQ_LAST(&pp->idle, workerhead);
+ if (w != NULL &&
+ (w->lastused < t_idle || pp->nthr > params->wthread_max)) {
+ VTAILQ_REMOVE(&pp->idle, w, list);
+ } else
+ w = NULL;
+ Lck_Unlock(&pp->mtx);
+
+ /* And give it a kiss on the cheek... */
+ if (w != NULL) {
+ pp->nthr--;
+ Lck_Lock(&pool_mtx);
+ VSC_C_main->threads--;
+ VSC_C_main->threads_destroyed++;
+ Lck_Unlock(&pool_mtx);
+ AZ(w->sp);
+ AZ(pthread_cond_signal(&w->cond));
}
}
- NEEDLESS_RETURN(NULL);
}
/*--------------------------------------------------------------------
- * Wait for another request
+ * Add a thread pool
*/
-void
-Pool_Wait(struct sess *sp)
+static struct pool *
+pool_mkpool(void)
{
+ struct pool *pp;
+ struct listen_sock *ls;
+ struct poolsock *ps;
+ pthread_condattr_t cv_attr;
- CHECK_OBJ_NOTNULL(sp, SESS_MAGIC);
- AZ(sp->obj);
- AZ(sp->vcl);
- assert(sp->fd >= 0);
- /*
- * Set nonblocking in the worker-thread, before passing to the
- * acceptor thread, to reduce syscall density of the latter.
- */
- if (VTCP_nonblocking(sp->fd))
- SES_Close(sp, "remote closed");
- waiter->pass(waiter_priv, sp);
+ ALLOC_OBJ(pp, POOL_MAGIC);
+ XXXAN(pp);
+ Lck_New(&pp->mtx, lck_wq);
+
+ VTAILQ_INIT(&pp->queue);
+ VTAILQ_INIT(&pp->idle);
+ VTAILQ_INIT(&pp->socks);
+ pp->sesspool = SES_NewPool(pp);
+ AN(pp->sesspool);
+
+ VTAILQ_FOREACH(ls, &heritage.socks, list) {
+ if (ls->sock < 0)
+ continue;
+ ALLOC_OBJ(ps, POOLSOCK_MAGIC);
+ XXXAN(ps);
+ ps->lsock = ls;
+ VTAILQ_INSERT_TAIL(&pp->socks, ps, list);
+ }
+
+ AZ(pthread_condattr_init(&cv_attr));
+ AZ(pthread_condattr_setclock(&cv_attr, CLOCK_MONOTONIC));
+ AZ(pthread_cond_init(&pp->herder_cond, &cv_attr));
+ AZ(pthread_condattr_destroy(&cv_attr));
+ Lck_New(&pp->herder_mtx, lck_herder);
+ AZ(pthread_create(&pp->herder_thr, NULL, pool_herder, pp));
+
+ return (pp);
+}
+
+/*--------------------------------------------------------------------
+ * This thread adjusts the number of pools to match the parameter.
+ *
+ */
+
+static void *
+pool_poolherder(void *priv)
+{
+ unsigned nwq;
+ VTAILQ_HEAD(,pool) pools = VTAILQ_HEAD_INITIALIZER(pools);
+ struct pool *pp;
+ uint64_t u;
+
+ THR_SetName("pool_herder");
+ (void)priv;
+
+ nwq = 0;
+ while (1) {
+ if (nwq < params->wthread_pools) {
+ pp = pool_mkpool();
+ if (pp != NULL) {
+ VTAILQ_INSERT_TAIL(&pools, pp, list);
+ VSC_C_main->pools++;
+ nwq++;
+ continue;
+ }
+ }
+ /* XXX: remove pools */
+ (void)sleep(1);
+ u = 0;
+ VTAILQ_FOREACH(pp, &pools, list)
+ u += pp->lqueue;
+ VSC_C_main->thread_queue_len = u;
+ }
}
/*--------------------------------------------------------------------*/
@@ -578,18 +550,10 @@ Pool_Wait(struct sess *sp)
void
Pool_Init(void)
{
- pthread_t tp;
-
- AZ(pthread_cond_init(&herder_cond, NULL));
- Lck_New(&herder_mtx, lck_herder);
waiter_priv = waiter->init();
-
- wrk_addpools(params->wthread_pools);
- AZ(pthread_create(&tp, NULL, wrk_herdtimer_thread, NULL));
- AZ(pthread_detach(tp));
- AZ(pthread_create(&tp, NULL, wrk_herder_thread, NULL));
- AZ(pthread_detach(tp));
+ Lck_New(&pool_mtx, lck_wq);
+ AZ(pthread_create(&thr_pool_herder, NULL, pool_poolherder, NULL));
}
/*--------------------------------------------------------------------*/
diff --git a/bin/varnishtest/tests/c00002.vtc b/bin/varnishtest/tests/c00002.vtc
index 88862b3..6914736 100644
--- a/bin/varnishtest/tests/c00002.vtc
+++ b/bin/varnishtest/tests/c00002.vtc
@@ -16,4 +16,4 @@ client c1 {
expect resp.status == 200
} -run
-varnish v1 -expect n_wrk == 8
+varnish v1 -expect threads == 8
diff --git a/include/vsc_fields.h b/include/vsc_fields.h
index fc81919..ea668cc 100644
--- a/include/vsc_fields.h
+++ b/include/vsc_fields.h
@@ -139,6 +139,68 @@ VSC_F(sessmem_limit, uint64_t, 1, 'c',
"Count of session memory allocations blocked by limit (max_sess)."
)
+/*---------------------------------------------------------------------
+ * Pools, threads, and sessions
+ * see: cache_pool.c
+ *
+ */
+
+VSC_F(pools, uint64_t, 1, 'g',
+ "Number of thread pools",
+ "Number of thread pools. See also param wthread_pools."
+ " NB: Presently pools cannot be removed once created."
+)
+
+VSC_F(threads, uint64_t, 1, 'g',
+ "Total number of threads",
+ "Number of threads in all pools."
+ " See also params thread_pools, thread_pool_min & thread_pool_max."
+)
+
+VSC_F(threads_limited, uint64_t, 1, 'c',
+ "Threads hit max",
+ "Number of times more threads were needed, but limit was reached"
+ " in a thread pool."
+ " See also param thread_pool_max."
+)
+
+VSC_F(threads_created, uint64_t, 1, 'c',
+ "Threads created",
+ "Total number of threads created in all pools."
+)
+
+VSC_F(threads_destroyed, uint64_t, 1, 'c',
+ "Threads destoryed",
+ "Total number of threads destroyed in all pools."
+)
+
+VSC_F(threads_failed, uint64_t, 1, 'c',
+ "Thread creation failed",
+ "Number of times creating a thread failed."
+ " See VSL::Debug for diagnostics."
+ " See also param thread_fail_delay."
+)
+
+VSC_F(thread_queue_len, uint64_t, 1, 'g',
+ "Length of session queue",
+ "Length of session queue waiting for threads."
+ " NB: Only updates once per second."
+ " See also param queue_max."
+)
+
+VSC_F(sess_queued, uint64_t, 1, 'c',
+ "Sessions queued for thread",
+ "Number of times session was queued waiting for a thread."
+ " See also param queue_max."
+)
+
+VSC_F(sess_dropped, uint64_t, 1, 'c',
+ "Sessions dropped for thread",
+ "Number of times session was dropped because the queue were too"
+ " long already."
+ " See also param queue_max."
+)
+
/*---------------------------------------------------------------------*/
VSC_F(n_sess_mem, uint64_t, 0, 'i', "N struct sess_mem", "")
@@ -150,14 +212,7 @@ VSC_F(n_objecthead, uint64_t, 1, 'i', "N struct objecthead", "")
VSC_F(n_waitinglist, uint64_t, 1, 'i', "N struct waitinglist", "")
VSC_F(n_vbc, uint64_t, 0, 'i', "N struct vbc", "")
-VSC_F(n_wrk, uint64_t, 0, 'i', "N worker threads", "")
-VSC_F(n_wrk_create, uint64_t, 0, 'a', "N worker threads created", "")
-VSC_F(n_wrk_failed, uint64_t, 0, 'a',
- "N worker threads not created", "")
-VSC_F(n_wrk_max, uint64_t, 0, 'a', "N worker threads limited", "")
-VSC_F(n_wrk_lqueue, uint64_t, 0, 'a', "work request queue length", "")
-VSC_F(n_wrk_queued, uint64_t, 0, 'a', "N queued work requests", "")
-VSC_F(n_wrk_drop, uint64_t, 0, 'a', "N dropped work requests", "")
+
VSC_F(n_backend, uint64_t, 0, 'i', "N backends", "")
VSC_F(n_expired, uint64_t, 0, 'i', "N expired objects", "")
More information about the varnish-commit
mailing list