lb_aux.cc

这是关于覆盖网的各种负载均衡算法的程序。好好研究

/* $Id: lb_aux.cc,v 1.17 2005/08/02 19:45:32 jonathan Exp $
 * Jonathan Ledlie, Harvard University.
 * Copyright 2005.  All rights reserved.
 */

#include "lb.h"

#ifdef DEBUG
bool debug;
#endif

FILE *traceFP = NULL;
static FILE *churnFP;

//next event action/round/id
int nextEventRound = -1;
char nextEventAction;
int nextEventNodeId;

//message records
int** queryMsg;
int** maintMsg;
int totalMsgCount = 0;

double lowerSlack;//下限缺省0.01，也可参数指定
double upperSlack;//上限缺省0.99，也可参数指定 

int cRound;//当前轮数
int idChoice;//k-choices 策略的k

int maxVsPerNode;//每个节点最大VS 数量

bool oracle;

//k-choices: dampening Strategy: 4 types 
bool dampeningLimitCrDel = false;
bool dampeningLimitIds = false;

// Note that this prints routed msgs whether or not they succeeded
void printLinks (PhysicalServer *ps, FILE *fp, bool doPrintLinks) {
  for (int i = 0; i < psCount; i++) {
    for (int j = 0; j < psCount; j++) {
      if (doPrintLinks) fprintf (fp, "%d -> %d %d %d\n", i, j, queryMsg[i][j], maintMsg[i][j]);
      ps[i].incrMaintMsgOut (maintMsg[i][j]);
      ps[j].incrMaintMsgIn (maintMsg[i][j]);
      ps[i].incrQueryMsgOut (queryMsg[i][j]);
      ps[j].incrQueryMsgIn (queryMsg[i][j]);
    }
  }
}

void printPs (PhysicalServer *ps,  FILE *fp) {
  for (int i = 0; i < psCount; i++) {
    ps[i].print (fp);
  }
}

//used in k-choices
double getTargetMismatch (PhysicalServer *ps) {
  double sum = 0.;
  int upCount = 0;
  for (int i = 0; i < psCount; i++) {
    if (ps[i].isAlive()) {
      sum += fabs (ps[i].getWorkPerRound()-ps[i].getUpperTarget());
      upCount++;
    }
  }
  return sum/(double)upCount;
}

void printVs (FILE *fp) {
  int count = 0;
  for (map<double,VirtualServer*>::iterator p = vServers.begin();
       p != vServers.end(); p++) {
	    count++;
	    map<double,VirtualServer*>::iterator nextStep = p;
	    nextStep++;
	    if (nextStep == vServers.end())
	      nextStep = vServers.begin();
	    double distance = keyDistance (p->first,nextStep->first);

	    fprintf (fp, "%d %d %f %f\n", count, p->second->getRootPs()->getId(),
		     p->first, distance);
	  }
}

void printArcs (PhysicalServer *ps, int psCount) {
  /*
  vector<double> arcs;
  int upCount = 0;
  for (int i = 0; i < psCount; i++) {
    //    debug = true;
    if (ps[i].isAlive()) {
      //      double wcr = ps[i].getWorkToCapacityRatio ();
      arcs.push_back (ps[i].getArcLength());
      //      arcs.push_back (wcr);
      upCount++;
    }
  }
  sort(arcs.begin(), arcs.end());  
  for (int i = 0; i < upCount; i++) {
    printf ("%d %f\n", i, arcs[i]);
  }
  */
}

//物理节点的构建
int initNodes (char *filename, PhysicalServer *&ps, int &initialVsPerNode) {
  int psCount = 0;
  FILE *fp;
  vector<int> capacity;
  if ((fp = fopen(filename, "r")) == NULL) {
    printf("%s: file open error.\n", filename);
    exit (-1);
  }

  // 从capacity file中读取各个物理节点数据node-id capacity
  // 文件数据格式: 节点顺序号,  能力值
  int id, cap;
  while (fscanf (fp, "%d %d\n", &id, &cap) > 0) {//id  cap节点能力
    if (capacity.size() <= id) {
      capacity.resize (id+1);
    }
    capacity[id] = cap;// store the capacity value 
    psCount++;   //物理节点计数
  }
  fclose (fp);


  //物理节点集合的构建
  ps = new PhysicalServer[psCount];//生成物理节点集合空间
  ASSERT (ps);
  //物理节点的初始化
  for (int i = 0; i < psCount; i++) {
    ps[i].setNode (i, (double)(capacity[i]));//各个物理节点初始化
  }

  //创建各个节点发送查询消息和维护消息计数数组
  queryMsg = new int* [psCount];
  maintMsg = new int* [psCount];

  for (int i = 0; i < psCount; i++) {
    queryMsg[i] = new int [psCount];
  }
  for (int i = 0; i < psCount; i++) {
    maintMsg[i] = new int [psCount];
  }
  
  for (int i = 0; i < psCount; i++) {
    for (int j = 0; j < psCount; j++) {
      queryMsg[i][j] = 0;
      maintMsg[i][j] = 0;
    }
  }


//参数用来定义finger 表的大小
  initStep (psCount/2 * initialVsPerNode);//初始化为所有的虚拟节点数量/2

  return psCount;
}

void deleteNodes (PhysicalServer *ps) {
  delete [] ps;

  for (int i = 0; i < psCount; i++) {
    int* deleteMe = queryMsg[i];
    delete [] deleteMe;
    deleteMe = maintMsg[i];
    delete [] deleteMe;
  }
  delete [] queryMsg;
  delete [] maintMsg;
}

void openChurnFile (char *filename) {
  if ((churnFP = fopen(filename, "r")) == NULL) {
    printf("%s: file open error.\n", filename);
    exit (-1);
  }
}



//该函数重要
//返回值为下列字符的含义
// ' ' 当前轮结束，将进入下一轮
// 'q' 整个事件结束
// nextEventAction 待处理的事件类型'b', 'd'

char nextEvent (int &nodeid) {
  int round;
  char action;
  int node;

  if (nextEventRound > cRound) { 
  	//确保当前轮完成
  	//将进入下一轮处理
    return ' ';
  }

  if (nextEventRound == cRound) {
  	//
    nextEventRound = -1;
    nodeid = nextEventNodeId;
    return nextEventAction;
  }

  if (fscanf (churnFP, "%d %c %d\n", &round, &action, &node) <= 0) {//
 	   return 'q';//异常读取扰动数据: 读取到文件末尾
  } 
  else {//读取上述三个参数
	 
	     //三个参数的合法性检查 
	    ASSERT (round >= cRound);//读取的轮数不小于当前轮数
	    if (node > psCount || node < 0) {//节点id 合法性
	      printf ("node %d, psCount %d, round %d\n", node, psCount, round);
	      ASSERT (node < psCount && node >= 0);    }

	    if (round == cRound) {//当前轮的事件
	      nodeid = node;
	      return action;
	    } 
	    else {//下一轮事件
	      nextEventRound = round;
	      nextEventAction = action;
	      nextEventNodeId = node;
	    }
  }
  return ' ';
}

void closeChurnFile () {
  // for some reason, fclose hangs.
  //  fclose (churnFP);
}

FILE* openOutputFile (char *prefix, char *suffix) {
  FILE *fp;
  char *filename;
  int fileLength = strlen(prefix)+strlen(suffix)+1;

  filename = new char [fileLength];
  memset (filename,0,fileLength);
  sprintf (filename, "%s%s", prefix, suffix);
  if ((fp = fopen(filename, "w")) == NULL) {
    printf("%s: file open error.\n", filename);
    exit (-1);
  }
  delete [] filename;
  return fp;
}


//less useful
int methodToCode (int idChoice, int initialVsPerNode, char activeLBmethod) {
  int b = 0;
  if (activeLBmethod == '-')
    b = 1;
  else if (activeLBmethod == 'k')
    b = 2;
  else if (activeLBmethod == 't')
    b = 3;
  else if (activeLBmethod == 'p')
    b = 4;
  return b;
}




const int MAX_QUERY_SUCCESS_LIST_SIZE = 20;

void initializePreviousQuerySuccess (deque<bool> &previousQuerySuccess) {
  for (int i = 0; i < MAX_QUERY_SUCCESS_LIST_SIZE; i++) {
    previousQuerySuccess.push_front (true);
  }
}

void addToPreviousQuerySuccess (deque<bool> &previousQuerySuccess,
                                bool routeSuccess) {
  previousQuerySuccess.pop_back();
  previousQuerySuccess.push_front (routeSuccess);
}

double avgOfPreviousQuerySuccess (deque<bool> &previousQuerySuccess,
                                  int sizeOfAvg) {
  double sum = 0.;
  ASSERT (sizeOfAvg <= MAX_QUERY_SUCCESS_LIST_SIZE);
  for (int i = 0; i < sizeOfAvg; i++) {
    if (previousQuerySuccess[i]) 
      sum += 1.;
  }
  return sum/sizeOfAvg;
}