common-agent.cc

这是P2P流媒体方案－NICE的实现源码

    		data->lid = htonl(ap->u.random_p.lid);
    		data->direction = htonl(ap->u.random_p.direction);
    		data->hop_count = htonl(ap->u.random_p.hop_count);
    		data->level = htonl(ap->u.random_p.level);

    		data->rand_id.agent_id = htonl(ap->u.random_p.rand_id.agent_id);
    		memcpy(&data->rand_id.agent_addr, &ap->u.random_p.rand_id.agent_addr, sizeof(struct sockaddr_in));
		
		
		//Time added
		memcpy(& data->send_time, & ap->u.random_p.send_time, sizeof(double));
  
    		assert((unsigned short) (5000 + ap->u.random_p.src_id.agent_id) == ntohs(ap->u.random_p.src_id.agent_addr.sin_port));
    		data->seq_no = htonl(ap->u.random_p.seq_no);
    		memcpy(data->data, ap->u.random_p.data, NUM_BUDDY_BITMAP*sizeof(char));
    		len += sizeof(struct query_random);
	}
	break;
	case (RETRANSMIT_REQUEST) : {
    		struct retransmit_pkt * data = (struct retransmit_pkt *) ((char *)payload + len);
    		data->src.agent_id = htonl(ap->u.retransmit_p.src.agent_id);
    		memcpy(&data->src.agent_addr, &ap->u.retransmit_p.src.agent_addr, sizeof(struct sockaddr_in));
    		assert((unsigned short) (5000 + ap->u.retransmit_p.src.agent_id) == ntohs(ap->u.retransmit_p.src.agent_addr.sin_port));
    		data->seq_no = htonl(ap->u.retransmit_p.seq_no);

		
		//Time added
		memcpy(& data->send_time, & ap->u.retransmit_p.send_time, sizeof(double));
  
    		memcpy(data->bitmap, ap->u.retransmit_p.bitmap, BITMAP_SIZE*sizeof(char));
    		len += sizeof(struct retransmit_pkt);
	}
	break;
//sunny
      
  case (PACKET_UNDEFINED) :
       
    printf("[Err] undefined packet\n");
    break; 
      
  default : 

    /* printf("[Err] undefined packet\n"); */
    break;

  }
  fflush(stdout);
  return len;
}

void commonAgent::free_app_packet(AppPacket *ap) {

  switch (ap->st) {

  case (CLUSTER_REFRESH) : {
    
    if (ap->u.clusterrefresh_p.hl_mbr_count > 0) {
      assert(ap->u.clusterrefresh_p.hl_agent_arr);
      free(ap->u.clusterrefresh_p.hl_agent_arr);
      ap->u.clusterrefresh_p.hl_agent_arr = NULL;
    }
    break; 
  }

  case (PING_RESPONSE) : {

    if (ap->u.pingresp_p.mbr_count > 0) {
      assert(ap->u.pingresp_p.agent_arr);
      free(ap->u.pingresp_p.agent_arr);
      ap->u.pingresp_p.agent_arr = NULL;
    }
  }
    break;

  case (PACKET_UNDEFINED) :
       
    printf("[Err] undefined packet\n");
    break; 

  case (PACKET_DATA) : {
    assert(ap->u.data_p.payload);
    //free(ap->u.data_p.payload);
  }
    break;

  default : 

    /* printf("undefined packet\n"); */
    break;
  }
  assert(ap);
  if(ap->st != PACKET_DATA) {
    delete ap;
    ap = NULL;
  }
  fflush(stdout);
  return;
}

/*
void commonAgent::handle_expired_timers() {
  
  int call_unmask_timer;

  call_unmask_timer = mask_timer();
  glob_handling_expired_timers = true;
  for (void *pos = expired_timer_list.GetHeadPosition();
       pos != NULL;
       expired_timer_list.GetNext(pos) ) {
    Timer *t = expired_timer_list.GetAt(pos);
    if (verify_and_remove_timer_instance(t->get_instance()) == true)
      t->Handler(true);
  }

  expired_timer_list.RemoveAll();
  glob_handling_expired_timers = false;
  if(call_unmask_timer == CALL_UNMASK_TIMER)
    unmask_timer();

  fflush(stdout);
  return;
}

void commonAgent::add_expired_timer(Timer * t) {

  expired_timer_list.Add(t, t->get_id());
  return;
}
*/



//PRM

int IdCache::getAgentId( int ni ) {
  /*
	if ( ni >= 0 && recent_nid == ni )
		return recent_aid;

	if ( ni >= 0 )
		for ( int i = 0; i < count; i++ ) {
			if ( nid[i] == ni ) {
				recent_aid = aid[i];
				recent_nid = nid[i];
				return aid[i];
			}
		}
  */
	if ( count <= 0 )
		return -1;

	int index = -1;
	if ( USE_CORRELATION ) {
		if ( ni == MISS_BASED_CORR && sum_miss > 0 ) {
			int rnum = rand() % sum_miss;
			for ( int i = 0 ; i < count; i++ ) {
				rnum -= miss_score[i];
				if ( rnum < 0 ) {
					index = i;
					break;
				}
			}
		}
		if ( ni == RECV_BASED_CORR && sum_recv > 0 ) {
			int rnum = rand() % sum_recv;
			for ( int i  = 0 ; i  < count; i ++ ) {
				rnum -= recv_score[i];
				if ( rnum < 0 ) {
					index = i;
					break;
				}
			}
		}
	}

	if ( index < 0 ) {
	 	index = rand() % count;
	}
	recent_aid = aid[index];
	
	//recent_nid = nid[index];
	
	return recent_aid;
}


//int IdCache::getNodeId( int ai ) {
void IdCache::getNodeId(int ai, struct sockaddr_in * nid) {

  //	if ( ai >= 0 && recent_aid == ai )
  //		return recent_nid;
  
  nid = NULL;

	if ( ai >= 0 )
		for ( int i = 0; i < count; i++ ) {
			if ( aid[i] == ai ) {
				recent_aid = aid[i];
				//recent_nid = nid[i];				
				//return nid[i];

				nid = &(addr[i]);
				return;
			}
		}

	if ( count <= 0 )
		return;

	int index = rand() % count;
	recent_aid = aid[index];
	
	//recent_nid = nid[index];
	
	//return recent_nid;
	nid = &(addr[index]);
	return;
}

int IdCache::addId( AppPacket *ap, int in_rscore, int in_mscore ) {
	query_random *ptr = &ap->u.random_p;
	
	//PRM
	//return addId( ptr->rand_id.agent_id, ptr->rand_id.node_id, ptr->level, in_rscore, in_mscore );
	return addId( ptr->rand_id.agent_id, &(ptr->rand_id.agent_addr), ptr->level, in_rscore, in_mscore );
}

//int IdCache::addId(int ai, int ni, int lvl, int in_rscore, int in_mscore ) {
int IdCache::addId(int ai, struct sockaddr_in * ni, int lvl, int in_rscore, int in_mscore){
		   int index = found(ai/*, ni*/);
	if ( index >= 0 ) {
		sum_recv -= recv_score[index];
		recv_score[index] = in_rscore;
		sum_recv += recv_score[index];

		sum_miss -= miss_score[index];
		miss_score[index] = in_mscore;
		sum_miss += miss_score[index];
	}
	//not found
	else if ( count < MAX_ID_CACHE ) {
		aid[count] = ai;
		
		//nid[count] = ni;
		//PRM
		memcpy(&(addr[count]), ni, sizeof(struct sockaddr_in));

		recv_score[count] = in_rscore;
		miss_score[count] = in_mscore;
		level[count] = 1 << (2 *lvl);

		sum_level += level[count];
		sum_recv += recv_score[count];
		sum_miss += miss_score[count];
		count++;
	}
	else {
		int index = getId();
		aid[index] = ai;
		//nid[index] = ni;
		//PRM
		memcpy(&(addr[index]), ni, sizeof(struct sockaddr_in));

		sum_level -= level[index];
		sum_recv -= recv_score[index];
		sum_miss -= miss_score[index];

		recv_score[count] = in_rscore;
		miss_score[count] = in_mscore;
		level[index] = 1 << (2 * lvl);

		sum_level += level[index];
		sum_recv += recv_score[index];
		sum_miss += miss_score[index];
	}

	return 1;
}

int IdCache::getId() {

	int i;
	for ( i = 0; i < count; i++ ) {
		if ( recv_score[i] == 0 && miss_score[i] == 0 )
			return i;
	}

	//fprintf(stderr, "non-zero id\n");
	return rand() % count;
}

int IdCache::found( int ai/*, int ni*/) {
	int i;
	for ( i = 0 ; i < count; i++ ) 
	  if ( aid[i] ==  ai /*&& nid[i] == ni */)
			return i;

	return -1;
}

void IdCache::dump() {
	printf("----------------------\n");
	printf("sum_recv: %d, sum_miss: %d\n", sum_recv, sum_miss);
	for ( int i = 0 ; i < count; i++ ) {
	  printf("\t%5d %5d %5d\n", aid[i], /*nid[i],*/ recv_score[i], miss_score[i]);
	}
}

////////////////////////////////////

void RxmitCache::clear() {
	memset( valid, 0, RXMIT_CACHE_SIZE ); 
}

int RxmitCache::lookup( int seq , double now) {
	int index = seq % RXMIT_CACHE_SIZE;
	if ( valid[index] && seq_no[index] == seq && timeout[index] > now )
		return 1;
	return 0;
}

int RxmitCache::insert( int seq, double rtt ) {

  printf(" The RTT time here in Retransmit_cache is %d %f\n", seq, rtt);

	assert ( rtt >= 0.0 );
	//	double now = Scheduler::Clock();
	double now = my_clock();

	if ( lookup( seq, now ) )
		return 1;

	int index = seq % RXMIT_CACHE_SIZE;
	valid[index] = 1;
	seq_no[index] = seq;
	timeout[index] = now + rtt * RXMIT_MULT;

	//fprintf(stderr, "%f %f %f\n", now, rtt, timeout[index] );

	return 0;
}
		
	
//slee