Implement "insert_pass" mode where we cache that an entity must be passed.

git-svn-id: http://www.varnish-cache.org/svn/trunk/varnish-cache@459 d4fa192b-c00b-0410-8231-f00ffab90ce4

bin/varnishd/cache.h

@@ -116,6 +116,8 @@ struct object {
 	unsigned		heap_idx;
 	unsigned		ban_seq;
 
+	unsigned		pass;
+
 	unsigned		response;
 
 	unsigned		valid;
@@ -260,6 +262,7 @@ extern pthread_mutex_t sessmtx;
 
 /* cache_pass.c */
 void PassSession(struct worker *w, struct sess *sp);
+void PassBody(struct worker *w, struct sess *sp);
 
 /* cache_pipe.c */
 void PipeSession(struct worker *w, struct sess *sp);

bin/varnishd/cache_center.c

@@ -84,6 +84,7 @@ cnt_done(struct worker *w, struct sess *sp)
 {
 	char *b;
 
+	(void)w;
 	if (http_GetHdr(sp->http, "Connection", &b) &&
 	    !strcmp(b, "close")) {
 		vca_close_session(sp, "Connection header");
@@ -168,11 +169,24 @@ cnt_fetch(struct worker *w, struct sess *sp)
 
 	if (sp->handling == VCL_RET_LOOKUP)
 		INCOMPL();
-	if (sp->handling == VCL_RET_PASS)
-		INCOMPL();
-	if (sp->handling == VCL_RET_INSERT_PASS)
-		INCOMPL();
+	if (sp->handling == VCL_RET_PASS) {
+		sp->obj->cacheable = 0;
+		HSH_Unbusy(sp->obj);
+		HSH_Deref(sp->obj);
+		sp->obj = NULL;
+		sp->step = STP_PASSBODY;
+		return;
+	}
+	if (sp->handling == VCL_RET_INSERT_PASS) {
+		sp->obj->pass = 1;
+		sp->obj->cacheable = 1;
+		HSH_Unbusy(sp->obj);
+		sp->obj = NULL;
+		sp->step = STP_PASSBODY;
+		return;
+	}
 	if (sp->handling == VCL_RET_INSERT) {
+		sp->obj->cacheable = 1;
 		FetchBody(w, sp);
 		sp->step = STP_DELIVER;
 		return;
@@ -225,15 +239,10 @@ cnt_hit(struct worker *w, struct sess *sp)
 {
 
 	VCL_hit_method(sp);
-	if (sp->handling == VCL_RET_LOOKUP)
-		INCOMPL();
-	if (sp->handling == VCL_RET_PASS) {
-		PassSession(w, sp);
-		sp->step = STP_DONE;
-		return;
-	}
-	if (sp->handling == VCL_RET_ERROR)
-		INCOMPL();
+
+	if (sp->handling == VCL_RET_DELIVER && sp->obj->pass)
+		sp->handling = VCL_RET_PASS;
+
 	if (sp->handling == VCL_RET_DELIVER) {
 		vca_write_obj(w, sp);
 		HSH_Deref(sp->obj);
@@ -241,6 +250,20 @@ cnt_hit(struct worker *w, struct sess *sp)
 		sp->step = STP_DONE;
 		return;
 	}
+	if (sp->handling == VCL_RET_PASS) {
+		HSH_Deref(sp->obj);
+		sp->obj = NULL;
+		PassSession(w, sp);
+		sp->step = STP_PASSBODY;
+		return;
+	}
+
+	if (sp->handling == VCL_RET_ERROR)
+		INCOMPL();
+
+	if (sp->handling == VCL_RET_LOOKUP)
+		INCOMPL();
+
 	INCOMPL();
 }
 
@@ -272,11 +295,17 @@ cnt_lookup(struct worker *w, struct sess *sp)
 	if (sp->obj->busy) {
 		VSL_stats->cache_miss++;
 		sp->step = STP_MISS;
-	} else {
-		VSL_stats->cache_hit++;
-		VSL(SLT_Hit, sp->fd, "%u", sp->obj->xid);
-		sp->step = STP_HIT;
+		return;
 	}
+	if (sp->obj->pass) {
+		VSL_stats->cache_hitpass++;
+		VSL(SLT_HitPass, sp->fd, "%u", sp->obj->xid);
+		sp->step = STP_HIT;
+		return;
+	} 
+	VSL_stats->cache_hit++;
+	VSL(SLT_Hit, sp->fd, "%u", sp->obj->xid);
+	sp->step = STP_HIT;
 }
 
 
@@ -330,7 +359,7 @@ cnt_miss(struct worker *w, struct sess *sp)
 		HSH_Deref(sp->obj);
 		sp->obj = 0;
 		PassSession(w, sp);
-		sp->step = STP_DONE;
+		sp->step = STP_PASSBODY;
 		return;
 	}
 	if (sp->handling == VCL_RET_LOOKUP)
@@ -363,7 +392,7 @@ cnt_pass(struct worker *w, struct sess *sp)
 {
 
 	PassSession(w, sp);
-	sp->step = STP_DONE;	/* XXX */
+	sp->step = STP_PASSBODY;
 }
 
 
@@ -381,7 +410,13 @@ DOT }
 DOT passbody -> DONE
  */
 
-static void cnt_passbody(struct worker *w, struct sess *sp) { (void)w; (void)sp; INCOMPL(); }
+static void
+cnt_passbody(struct worker *w, struct sess *sp)
+{
+	PassBody(w, sp);
+	sp->step = STP_DONE;
+	return;
+}
 
 
 /*--------------------------------------------------------------------
@@ -493,7 +528,11 @@ CNT_Session(struct worker *w, struct sess *sp)
 
 	for (sp->step = STP_RECV; sp->step != STP_DONE; ) {
 		switch (sp->step) {
-#define STEP(l,u) case STP_##u: cnt_##l(w, sp); break;
+#define STEP(l,u) \
+		case STP_##u: \
+			VSL(SLT_Debug, sp->fd, "State " #u); \
+			cnt_##l(w, sp); \
+			break;
 #include "steps.h"
 #undef STEP
 		default:	INCOMPL();

bin/varnishd/cache_fetch.c

@@ -265,9 +265,9 @@ FetchBody(struct worker *w, struct sess *sp)
 	else
 		VBE_RecycleFd(vc);
 
-	if (sp->obj->cacheable)
-		EXP_Insert(sp->obj);
 	HSH_Unbusy(sp->obj);
+	/* Hold on to the reference count, it's not released until
+	 * expiry */
 	if (!sp->obj->cacheable)
 		HSH_Deref(sp->obj);
 	return (0);

bin/varnishd/cache_hash.c

@@ -111,6 +111,8 @@ void
 HSH_Unbusy(struct object *o)
 {
 
+	if (o->cacheable)
+		EXP_Insert(o);
 	AZ(pthread_mutex_lock(&o->objhead->mtx));
 	o->busy = 0;
 	AZ(pthread_mutex_unlock(&o->objhead->mtx));

bin/varnishd/cache_pass.c

@@ -144,33 +144,19 @@ pass_chunked(struct sess *sp, int fd, struct http *hp)
 
 
 /*--------------------------------------------------------------------*/
+
 void
-PassSession(struct worker *w, struct sess *sp)
+PassBody(struct worker *w, struct sess *sp)
 {
-	int i;
 	struct vbe_conn *vc;
 	struct http *hp;
 	char *b;
 	int cls;
 
-	vc = VBE_GetFd(sp->backend, sp->xid);
+	hp = sp->bkd_http;
+	assert(hp != NULL);
+	vc = sp->vbc;
 	assert(vc != NULL);
-	VSL(SLT_Backend, sp->fd, "%d %s", vc->fd, sp->backend->vcl_name);
-
-	http_BuildSbuf(vc->fd, Build_Pass, w->sb, sp->http);
-	i = write(vc->fd, sbuf_data(w->sb), sbuf_len(w->sb));
-	assert(i == sbuf_len(w->sb));
-
-	/* XXX: copy any contents */
-
-	/*
-	 * XXX: It might be cheaper to avoid the event_engine and simply
-	 * XXX: read(2) the header
-	 */
-	hp = vc->http;
-	http_RecvHead(hp, vc->fd, w->eb, NULL, NULL);
-	event_base_loop(w->eb, 0);
-	http_DissectResponse(hp, vc->fd);
 
 	http_BuildSbuf(sp->fd, Build_Reply, w->sb, hp);
 	vca_write(sp, sbuf_data(w->sb), sbuf_len(w->sb));
@@ -194,3 +180,34 @@ PassSession(struct worker *w, struct sess *sp)
 	else
 		VBE_RecycleFd(vc);
 }
+
+/*--------------------------------------------------------------------*/
+void
+PassSession(struct worker *w, struct sess *sp)
+{
+	int i;
+	struct vbe_conn *vc;
+	struct http *hp;
+
+	vc = VBE_GetFd(sp->backend, sp->xid);
+	assert(vc != NULL);
+	VSL(SLT_Backend, sp->fd, "%d %s", vc->fd, sp->backend->vcl_name);
+
+	http_BuildSbuf(vc->fd, Build_Pass, w->sb, sp->http);
+	i = write(vc->fd, sbuf_data(w->sb), sbuf_len(w->sb));
+	assert(i == sbuf_len(w->sb));
+
+	/* XXX: copy any contents */
+
+	/*
+	 * XXX: It might be cheaper to avoid the event_engine and simply
+	 * XXX: read(2) the header
+	 */
+	hp = vc->http;
+	http_RecvHead(hp, vc->fd, w->eb, NULL, NULL);
+	event_base_loop(w->eb, 0);
+	http_DissectResponse(hp, vc->fd);
+
+	sp->bkd_http = hp;
+	sp->vbc = vc;
+}

include/shmlog_tags.h

@@ -35,6 +35,7 @@ SLTM(VCL_trace)
 SLTM(VCL_return)
 SLTM(XID)
 SLTM(Hit)
+SLTM(HitPass)
 SLTM(ExpBan)
 SLTM(ExpPick)
 SLTM(ExpKill)

include/stat_field.h

@@ -4,6 +4,7 @@ MAC_STAT(client_conn,		uint64_t, "u", "Client connections accepted")
 MAC_STAT(client_req,		uint64_t, "u", "Client requests received")
 
 MAC_STAT(cache_hit,		uint64_t, "u", "Cache hits")
+MAC_STAT(cache_hitpass,		uint64_t, "u", "Cache hits for pass")
 MAC_STAT(cache_miss,		uint64_t, "u", "Cache misses")
 
 MAC_STAT(backend_conn,		uint64_t, "u", "Backend connections initiated")