Add VBO_waitlen()

bin/varnishd/cache/cache.h

@@ -796,7 +796,8 @@ void VBO_Init(void);
 struct busyobj *VBO_GetBusyObj(struct worker *, struct req *);
 void VBO_DerefBusyObj(struct worker *wrk, struct busyobj **busyobj);
 void VBO_Free(struct busyobj **vbo);
-void VBO_extend(const struct busyobj *, ssize_t);
+void VBO_extend(struct busyobj *, ssize_t);
+ssize_t VBO_waitlen(struct busyobj *bo, ssize_t l);
 void VBO_setstate(struct busyobj *bo, enum busyobj_state_e next);
 void VBO_waitstate(struct busyobj *bo, enum busyobj_state_e want);
 

bin/varnishd/cache/cache_busyobj.c

@@ -212,7 +212,7 @@ VBO_DerefBusyObj(struct worker *wrk, struct busyobj **pbo)
 }
 
 void
-VBO_extend(const struct busyobj *bo, ssize_t l)
+VBO_extend(struct busyobj *bo, ssize_t l)
 {
 
 	CHECK_OBJ_NOTNULL(bo, BUSYOBJ_MAGIC);
@@ -220,7 +220,25 @@ VBO_extend(const struct busyobj *bo, ssize_t l)
 	if (l == 0)
 		return;
 	assert(l > 0);
+	Lck_Lock(&bo->mtx);
 	bo->fetch_obj->len += l;
+	AZ(pthread_cond_signal(&bo->cond));
+	Lck_Unlock(&bo->mtx);
+}
+
+ssize_t
+VBO_waitlen(struct busyobj *bo, ssize_t l)
+{
+	Lck_Lock(&bo->mtx);
+	while (1) {
+		if (bo->fetch_obj->len > l || bo->state >= BOS_FINISHED) {
+			l = bo->fetch_obj->len;
+			break;
+		}
+		(void)Lck_CondWait(&bo->cond, &bo->mtx, NULL);
+	}
+	Lck_Unlock(&bo->mtx);
+	return (l);
 }
 
 void

bin/varnishd/cache/cache_http1_fetch.c

@@ -393,8 +393,10 @@ V1F_fetch_body(struct worker *wrk, struct busyobj *bo)
 		if (st->len == 0) {
 			VTAILQ_REMOVE(&bo->fetch_obj->store, st, list);
 			STV_free(st);
-		} else if (st->len < st->space)
+		} else if (st->len < st->space) {
+			/* XXX: is this safe under streaming ? */
 			STV_trim(st, st->len, 1);
+		}
 	}
 
 	bo->vfp = NULL;