media-app.cc

柯老师网站上找到的

// to the lowest layer, i.e. it fills the triangle in a bottom-up
// starting from the base layer. We use this routine instead of bufOpt,
// for all cases during filling phase. Allocation based on diagonal strips
//
double QA::bufOptScen1(int layer, int layers, double currrate, 
		       double slope, int backoffs)
{
	double smallt, larget, side, rate;
  
	if (backoffs < 0) {
		panic("# ERROR: backoff: %d in bufOptScen1\n", 
		      backoffs);
	}
	rate = currrate/pow(2,backoffs);
	side = LAYERBW_*layers - (rate + layer*LAYERBW_);
	if (side <= 0.0) 
		return(0.0);
	larget = BufNeed(side, slope);
	side = LAYERBW_*layers - (rate + (layer+1)*LAYERBW_);
	if (side < 0.0) 
		side = 0.0;
	smallt = BufNeed(side, slope);

	return (larget-smallt);
}

//
// This routine calculate optimal buffer distribution for a layer
// in scenario 2 based on the 
// 1) current rate, 2) no of layers, 3) no of backoffs
//
// Jan 28, 99bufOptScen1(layer, layers, currrate, slope, backoffs)
//
double QA::bufOptScen2(int layer, int layers, double currrate, 
		       double slope, int backoffs)
{
	double bufopt = 0.0;
	int bmin, done;

	if(backoffs < 0) {
		panic("# ERROR: backoff: %d in bufOptScen2\n", 
			backoffs);
	}
	if ((currrate/pow(2,backoffs)) >= layers*LAYERBW_)
		return(0.0);

	bmin = 0;
	done = 0;
	while ((!done) && bmin<=backoffs) {
		if(currrate/pow(2,bmin) >= LAYERBW_*layers)
			bmin++;
		else 
			done++;
	}
	// buf required for the first triangle
	// we could have dec bmin and go for 1 backoff as well
	bufopt = bufOptScen1(layer, layers, currrate/pow(2,bmin), slope, 0);
  
	// remaining sequential backoffs
	bufopt += (backoffs - bmin)*BufNeed(layers*LAYERBW_/2, slope);
	return(bufopt);
}


//
// This routine returns the optimal distribution of requested-to-drained
// buffer across active layers based on:
// 1) curr rate, 2) curr drain distr(FinalDrainArry), etc
// NOTE, the caller must update FinalDrainArray from 
// 
// Jan 29, 99
//
// DrainArr: return value, used as an incremental chaneg for 
//   FinalDrainArray
// bufAvail:  current buffer_ state
void QA::drain_buf(double* DrainArr, double bufToDrain, 
		   double* FinalDrainArray, double* bufAvail, 
		   int layers, double rate, double srtt)
{
	double bufReq1, bufReq2, bufs1[MAX_LAYER], bufs2[MAX_LAYER], slope, 
		extra, targetArr[MAX_LAYER], maxDrainRemain;
	int bs1, bs2, l;

	slope = seg_size_/srtt;
	bs1 = MAXBKOFF_ + 1;
	bs2 = MAXBKOFF_ + 1;
	bufReq1 = bufReq2 = 0;
	for(l=0; l<layers; l++){
		bufReq1 += bufOptScen1(l, layers, rate, slope, bs1);
		bufReq2 += bufOptScen2(l, layers, rate, slope, bs2);
	} 

	for(l=0; l<MAX_LAYER; l++){
		bufs1[l] = 0;
		bufs2[l] = 0;
		DrainArr[l] = 0.0;
	}

	while(bufReq1 > TotalBuf(layers, bufAvail)){
		bufReq1 = 0;
		bs1--;
		for(l=0; l<layers; l++){
			bufs1[l] = bufOptScen1(l, layers, rate, slope, bs1);
			bufReq1 += bufs1[l];
		} 
	}
  
	while(bufReq2 > TotalBuf(layers, bufAvail)){
		bufReq2 = 0;
		bs2--;
		for(l=0; l<layers; l++){
			bufs2[l] =  bufOptScen2(l, layers, rate, slope, bs2);
			bufReq2 += bufs2[l];
		} 
	}

	if (bufReq1 >= bufReq2) {
		// drain toward last optimal scenario 1
		for (l=layers-1; l>=0; l--){
			// we try to drain the maximum amount from
			// min no of highest layers
			// note that there is a limit on total draining
			// from a layer
			maxDrainRemain = srtt*LAYERBW_ - FinalDrainArray[l];
			if ((bufAvail[l] > bufs1[l] + maxDrainRemain) &&
			    (bufToDrain >= maxDrainRemain)) {
				DrainArr[l] = maxDrainRemain;
				bufToDrain -= maxDrainRemain;
			} else {
				if(bufAvail[l] > bufs1[l] + maxDrainRemain){
					DrainArr[l] = bufToDrain;
					bufToDrain = 0.0;
				} else {
					DrainArr[l] = bufAvail[l] - bufs1[l];
					bufToDrain -= bufAvail[l] - bufs1[l];

					/* for debug */
					if(DrainArr[l] < 0.0){
// 	    panic("# ERROR, DrainArr[%d]: %.2f, bufAvail: %.2f, bufs1: %.2f\n",
// 		  l, DrainArr[l], bufAvail[l], bufs1[l]);
						DrainArr[l] = 0.0;
					}
				}
			}
			if(bufToDrain == 0.0)
				return;
		}
		return;
	} else {   /* if (bufReq1 >= bufReq2) */

		// Drain towards he last optima scenario 2 
		// We're draining - don't care about the upper bound on 
		// scenario 2.
		// Have to calculate all the layers together to get this max 
		// thing to work
		extra = 0.0;
		// Calculate the extra buffering 
		for (l=0; l<layers; l++) {
			if(bufs1[l] > bufs2[l])
				extra += bufs1[l] - bufs2[l];
		}

		for (l=layers-1; l>=0; l--)
			if(bufs1[l] >= bufs2[l])
				targetArr[l] = bufs1[l];
			else
				if (bufs2[l] - bufs1[l] >= extra) {
					targetArr[l] = bufs2[l] - extra;
					extra = 0;
				} else {
					// there is enough extra to compenstae the dif
					if (extra > 0) {
						targetArr[l] = bufs2[l];
						extra -= bufs2[l] - bufs1[l];
					} else 
						panic("# ERROR Should not \
reach here, extra: %.2f, bufs2: %.2f, bufs1: %.2f, L%d\n", 
						extra, bufs2[l], bufs1[l], l);
				}
	} /* end of if (bufReq1 >= bufReq2) */
   
	// drain toward last optimal scenario 2
	for (l=layers-1; l>=0; l--) {
		// we try to drain the maximum amount from
		// min no of highest layers
		// note that there is a limit on total draining
		// from a layer
		maxDrainRemain = srtt*LAYERBW_ - FinalDrainArray[l];
		if ((bufAvail[l] > targetArr[l] + maxDrainRemain) &&
		    (bufToDrain >= maxDrainRemain)) {
			DrainArr[l] = maxDrainRemain;
			bufToDrain -= maxDrainRemain;
		} else {
			if(bufAvail[l] > targetArr[l] + maxDrainRemain){
				DrainArr[l] = bufToDrain;
				bufToDrain = 0.0;
			} else {
				DrainArr[l] = bufAvail[l] - targetArr[l];
				bufToDrain -= bufAvail[l] - targetArr[l];
	
				// for debug 
				if (DrainArr[l] < 0.0) {
// 	  panic("# ERROR, DrainArr[%d]: %.2f, bufAvail: %.2f, bufs1: %.2f\n",
// 		l, DrainArr[l], bufAvail[l], bufs1[l]);
					DrainArr[l] = 0;
				}
			}
		}
		if (bufToDrain == 0.0)
			return;
	} /* end of for */
	return;
}


//
// This routine calculate an optimal distribution of a given 
// amount of buffered data to drain.
// the main algorithm is in drain_buf() and this one mainly init
// the input and calls that routine ad then update FinalDrainArray,
// based on its old value and return value for DrainArr.
//
// FinalDrainArray: output
// FinalBuffer: output, expected buf state at the end of the interval
void QA::DrainPacket(double bufToDrain, double* FinalDrainArray, int layers,
		     double rate, double srtt, double* FinalBuffer)
{
	double DrainArr[MAX_LAYER],  bufAvail[MAX_LAYER], TotBufAvail;
	int l,cnt;
  
	for(l=0; l<MAX_LAYER; l++){
		FinalDrainArray[l] = 0.0;
		bufAvail[l] = buffer_[l];
	}

	TotBufAvail = TotalBuf(layers, bufAvail);
	cnt = 0;
	while ((bufToDrain > 0) && (cnt < 10)) {
		// debug("bufToDrain%d: %.2f\n", cnt, bufToDrain);
		drain_buf(DrainArr, bufToDrain, FinalDrainArray, bufAvail, 
			  layers, rate, srtt);
		
		for(l=0; l<layers; l++){
			bufToDrain -= DrainArr[l];
			TotBufAvail -= DrainArr[l];
			FinalDrainArray[l] += DrainArr[l];
			bufAvail[l] -= DrainArr[l];
			FinalBuffer[l] = buffer_[l] - FinalDrainArray[l];
		}
		cnt++;
	}
}

void QA::check_availability(int layer, const MediaSegment& s) 
{
	int dummy; 
	MediaRequest p(MEDIAREQ_CHECKSEG);
	p.set_name(page_);
	p.set_layer(layer);
	p.set_st(s.start());
	p.set_et(s.end());
	p.set_app(this);
	// Ask cache/server to do prefetching if necessary.
	target()->get_data(dummy, &p);
}

/* 
 * This routine is called once every SRTT to drain some data from
 * recv's buffer and src's image from recv's buf.
 */
void QA::DrainBuffers()
{
	int i, j, layers = 0;
	Scheduler& s = Scheduler::instance();
	double now = s.clock();
	// interval since last drain
	double interval = now - playTime_;
	playTime_ = now;  // update playTime

	if ((layers > 1) && (playing_[0] != 1)) {
		panic("ERROR in DrainBuffer: layers>0 but L0 isn't playing\n");
	}

	// Updating playout offset, but do nothing if we are in the initial
	// startup filling phase! This offset measures the playing progress
	// of the client side. It is actually the playing offset of the lowest
	// layer.

	// This is the real amount of data to be drained from layers
	int todrain[MAX_LAYER]; 
	// Expected offset of base layer after draining, without considering 
	// holes in data. This has to be satisfied, otherwise base layer will
	// be dropped and an error condition will be raised.
	poffset_ += (int)floor(interval*LAYERBW_+0.5);

	// Started from MAX_LAYER to make debugging easier
	for (i = MAX_LAYER-1; i >= 0; i--) {
		// If this layer is not being played, don't drain anything
		if (sending_[i] == 0) {
			todrain[i] = 0;
			drained_[i] = 0.0;
			continue; 
		}
		todrain[i] = outlist_[i].evict_head_offset(poffset_);
		assert(todrain[i] >= 0);
		buffer_[i] -= todrain[i];
		// A buffer must have more than one byte
		if ((int)buffer_[i] <= 0) {
			debug("Buffer %d ran dry: %.2f after draining, DROP\n",
			      i, buffer_[i]);
			playing_[i] = 0;
			sending_[i] = 0;
			buffer_[i] = 0;
	    
			/* Drop all higher layers if they still have data */
			for (j = i+1; j < MAX_LAYER; j++)
				if (sending_[j] == 1) {
/*
 					panic("# ERROR: layer %d \
is playing with %.2f buf but layer %d ran dry with %.2f buf\n",
 					      j, buffer_[j], i, buffer_[i]);
*/
 					debug("# DROP layer %d: it \
is playing with %.2f buf but layer %d ran dry with %.2f buf\n",
 					      j, buffer_[j], i, buffer_[i]);
					sending_[j] = 0;
					playing_[j] = 0;
					buffer_[j] = 0;
				}
			// We don't need to set it to -1. The old address 
			// will be used to see if we are sending old data if 
			// that later is added again
			//
			// XXX Where is this -1 mark ever used????
			// data_[i].set_start(-1); // drop layer i
		} else {
			// Prefetch for this layer. Round to whole segment
			int st = (int)floor((poffset_+pref_srtt_*LAYERBW_)
					    /seg_size_+0.5)*seg_size_;
			int et = (int)floor((poffset_+(pref_srtt_+interval)*
					   LAYERBW_)/seg_size_+0.5)*seg_size_;
			if (et > pref_[i]) {
				pref_[i] = et;
				MediaSegment s(st, et);
				check_availability(i, s);
			}
		}
	} /* end of for */
}

// This routine dumps info into a file
// format of each line is as follows:
//  time tot-rate avg-rate per-layer-bw[MAXLAYER] tot-bw drain-rate[MAXLAYER] 
//  & cumulative-buffer[MAXLAYER] & no-of-layers 
// ADDED: use the old value of SRTT for bw/etc estimation !!! Jan 26
// XXX: need to be more compressed add more hooks to for ctrling from  
// tcl level
void QA::DumpInfo(double t, double last_t, double rate, 
		  double avgrate, double srtt)
{
#define MAXLEN 2000
	int i,j;
	char s1[MAXLEN], s2[MAXLEN], tmp[MAXLEN];
	static double last_srtt = 0, t1,t2 = 0;
#undef MAXLEN

	double  tot_bw = 0.0, interval, diff;
// 	if(rate > 1000000.0){
// 		debug("WARNING rate: %f is too large\n", rate);
// 	}
	
	interval = t - last_t ;
  
	if((t2 != last_t) && (t2 > 0)){
		diff = interval - last_srtt;
		if ((diff > 0.001) || (diff < -0.001)) {
			if (last_t == 0) 
				// Startup phase
				return;
/*
 			debug("WARNING: last_srtt: %.4f != \
interval: %.4f, diff: %f t1: %f, t2: %f, last_t: %f, t: %f\n",
 				last_srtt, interval, diff, t1, t2, last_t, t);
*/
			//abort();
		}
	} else 
		/* for the first call to init */
		last_srtt = srtt;

	t1 = last_t;
	t2 = t;

	if (interval <= 0.0) {
		panic("# ERROR interval is negative\n");
	}

	sprintf(s1, " %.2f %.2f %.2f X", last_t, rate, avgrate); 
	sprintf(s2, " %.2f %.2f %.2f X", t, rate, avgrate);
	
	j = 0;
	for (i = 0; i < MAX_LAYER; i++) 
	  //if (playing_[i] == 1)
	  if (sending_[i] == 1)
	    j++;
	//no of layers being playback
	sprintf(tmp, " %d", j*LAYERBW_);
	strcat(s1, tmp);
	strcat(s2, tmp);

	for (i = 0; i < MAX_LAYER; i++) {
		sprintf(tmp, " %.2g ", (bw_[i]/interval)+i*10000.0);
		strcat(s1,tmp);
		strcat(s2,tmp);

		tot_bw += bw_[i]/interval;
		bw_[i] = 0;
	}

	sprintf(tmp, " %.2f X", tot_bw );
	strcat(s1,tmp);
	strcat(s2,tmp);
  
	j = 0;
	for (i = 0; i < MAX_LAYER; i++) {
	  //if (playing_[i] == 1) {
	  if(sending_[i] ==1){
	    j++;
	    // drained_[] can be neg when allocated buf for this 
	    // layer is more than consumed data
	    if (drained_[i] < 0.0) {
	      // this means that this layer was drained 
	      // with max rate
	      drained_[i] = 0.0;
	    } 
	    // XXX, we could have used interval*LAYERBW_ - bw_[i]
	    // that was certainly better
	    sprintf(tmp, " %.2f ", 
		    (drained_[i]/interval)+i*10000.0);
	    strcat(s1,tmp);
	    strcat(s2,tmp);
	    // Note that drained[] shows the amount of data that 
	    // is used from buffered data, i.e. rd[i]
	    // This must be srtt instead of interval because this 
	    // is for next dumping.
	    drained_[i]=srtt*LAYERBW_;
	  } else {
	    sprintf(tmp, " %.2f ", i*10000.0);
	    strcat(s1,tmp);
	    strcat(s2,tmp);
	    drained_[i] = 0.0;
	  }
	}

	for (i=0;i<MAX_LAYER;i++) {
		sprintf(tmp, " %.2f", buffer_[i]+i*10000);
		strcat(s1,tmp);
		strcat(s2,tmp);
	}
	
	log("QA %s \n", s1);
	log("QA %s \n", s2);
	fflush(stdout);
}

// This routine models draining of buffers at the recv
// it periodically updates state of buffers
// Ir must be called once and then it reschedules itself
// it is first called after playout is started!
void QA::UpdateState()
{
	double last_ptime = playTime_; // Last time to drain buffer
	DrainBuffers();
	DumpInfo(Scheduler::instance().clock(), last_ptime, 
		 rate(), avgrate_, rap()->srtt());
}