chord.mac

这是一个著名的应用层组播中间件的源码

  joined recv update_pred {
    neighbor_finger *predent = neighbor_random( pred);

    // check to see if sending node should be our predecessor
    if (isinrange( field(id), predent->hashf, myhash)) {
      sprintf(trace_buf_, "change pred: old %.8x new %.8x\n", predent->hashf, field(id));
      trace_print();
      neighbor_remove( pred, predent->ipaddr);
      neighbor_add( pred, from);
      predent->hashf = field(id);
      upcall_notify(pred,NBR_TYPE_PEER);
    }
  }


  joined timer fix_fingers {
    // responsible for lazily correcting route (finger) table entries
    int myrand, start, powow, nextguy;
    neighbor_finger *myent, *predent;
    myent = neighbor_random(myfinger[0]);
    predent = neighbor_random(pred);
    if (predent->ipaddr == me || myent->ipaddr == me) 
      return; // if i am alone in this ring, get out

    myrand = randint(BITS-1) + 1; // select a random finger entry
    start = (int)pow(2, myrand);
    start = start + myhash; // the start address for this entry
    // route a find pred toward this start address
    nextguy = nexthop(start);
    route_find_pred( nextguy, start, me, 0, 0, -1); 

    powow = myhash - start; // go backwards in the ring to correct others
    // route an update others message toward them
    nextguy = nexthop(powow);
    route_update_others( nextguy, powow, myhash, me, 0, 0, -1);
  }

  joined recv find_pred_reply {
    // when attempting to fix my finger table, I will get replies from nodes
    // to consider
    register_knowledge(field(succ), field(succ_id)); // update fingers
  }

  joined recv update_others {
    int sendupdate, nextguy;
    neighbor_finger *predent, *myent;

    sendupdate = 0;  // don't send an update unless we change a finger entry
    myent = neighbor_random(myfinger[0]); // successor
    predent = neighbor_random(pred); // predecessor
      
    if (isinrange(field(id), myhash, myent->hashf)) { // final destination
      sendupdate = register_knowledge(field(who), field(node));
      if (sendupdate == 1 && predent->ipaddr != me 
	  && predent->ipaddr != field(who)) {
        route_update_ft(predent->ipaddr, field(node), field(who), 0, 0, -1);
      }
      return;
    }

    nextguy = nexthop (field(id));
    if (nextguy != me) { // route the message
      route_update_others(nextguy, field(id), field(node), field(who), 0, 0, -1);
    }
  }

  joined recv update_ft {
    int mybit, loopbit, sendupdate, powow;
    neighbor_finger *myent, *predent;

    sendupdate = register_knowledge(field(who), field(node));
    if (sendupdate == 1 && predent->ipaddr != me && predent->ipaddr != field(who)) { // only propagate if my state actually changed
      route_update_ft( predent->ipaddr, field(node), field(who), 0, 0, -1);
    }
  }

  any recv mapping {
    map.insert(field(low_hash), field(high_hash), field(map_ip));  
  }

  joined recv data {
    routedata(field(sender_hash), field(sender_ip), field(receiver_hash), size, field(priority), msg, 0);
  }

  joined recv data_path_req {
    routedata(field(sender_hash), field(sender_ip), field(receiver_hash), size, field(priority), msg, 1);
  }

  API route {
    routedata(hashof(me), me, dest, size, transport, msg, 0);
  }

  API routeIP {
    routedata(hashof(me), me, dest, size, transport, msg, 1);
  }

  any timer printer {
    dump_state();
  }

  API downcall_ext {
    switch(operation) {
      case TEST_OWNER:  
        {
          test_owner_arg *targ = (test_owner_arg*)arg;
          neighbor_finger *predent;
          predent = neighbor_random(pred);
          return isinrangeright(targ->nodeId,predent->hashf,myhash);
        }
      default: 
        { //downcall not supported
          return -1;
        }
    }
    return 0;
  }

  API error {
    neighbor_finger *myent, *myent2;
    int loopbit;
    sprintf(trace_buf_, "Neighbor %.8x died!\n", neighbor);
    trace_print();

    if (neighbor_query(pred, neighbor)) { // predecessor died
      neighbor_remove(pred, neighbor);
      myent = neighbor_random(myfinger[BITS-1]);
      neighbor_add(pred, myent->ipaddr);
      myent2 = neighbor_random(pred);
      myent2->hashf = myent->hashf;
      upcall_notify(pred,NBR_TYPE_PEER);
    }

    for (loopbit = BITS-1; loopbit>=1; loopbit--) {
      if (neighbor_query(myfinger[loopbit], neighbor)) {
        myent = neighbor_random(myfinger[loopbit-1]);
        neighbor_remove(myfinger[loopbit], neighbor);
        neighbor_add(myfinger[loopbit], myent->ipaddr);
      }
    }
    if (neighbor_query(myfinger[0], neighbor)) { // successor diede
      myent = neighbor_random(myfinger[0]);
      int old_start = myent->start;
      neighbor_remove(myfinger[0], neighbor);
      if (neighbor_size (successors_succ))
        myent = neighbor_random(successors_succ);
      else
        myent = neighbor_random(myfinger[1]);
      neighbor_add(myfinger[0], myent->ipaddr);
      myent2 = neighbor_random(myfinger[0]);
      myent2->start = old_start;
      myent2->hashf = myent->hashf;
    }
  }
}

routines {
  int nexthop (int target) {
    int i,j,done;
    neighbor_finger *predent, *myent, *myent2;

    done = 0; // haven't found the next hop yet
    // first check if the target is beyond last finger
    j = 0;
    i = BITS-1;
    myent = neighbor_random(myfinger[i]);
    myent2 = neighbor_random(myfinger[j]);
    if (isinrange( target, myent->start, myent2->start )) {
      done = 1;
    }
    for (j = BITS-1; j > 0; j--) { // loop thru fingers to find right one
      if (done == 0) {
        i = j - 1;
        myent = neighbor_random(myfinger[i]);
        myent2 = neighbor_random(myfinger[j]);
        if (isinrange( target, myent->start, myent2->start )) {
          done = 1;
        }
      }
    }
    done = 0; // reset flag
    for (j = i; j >= 0; j--) { // backs up to make sure that we forward to 
      // a node strictly before target
      if (isinrange( myent->hashf, myent->start, target )) {
        if (myent->start != target) {
          done = 1;
        }
      }
      if (myent->ipaddr == me || done == 0) { // still looking
	// note that I should not forward to myself (lightly populated case)
        myent = neighbor_random(myfinger[j]);
      }
    }
    return myent->ipaddr;  // could return me if i am the only node
  }

  int register_knowledge (int target_ip, int target_hash) {
    // updates the appropriate finger table entry 
    // returns 1 if an update occurred
    int mybit, loopbit, powow;
    int change=0;
    neighbor_finger *predent, *myent, *myent2;
    mybit = BITS-1;
    for (loopbit = mybit; loopbit >= 0; loopbit--) {
      myent = neighbor_random(myfinger[loopbit]);
      powow = myent->start;
      if (isinrange(target_hash, powow, myent->hashf)) {
	sprintf(trace_buf_, "change finger %d: old %.8x new %.8x start %.8x\n", loopbit, myent->hashf, target_hash, powow);
	trace_print();
	change=1;
	neighbor_remove( myfinger[loopbit], myent->ipaddr );
	neighbor_add( myfinger[loopbit], target_ip);
	myent = neighbor_random( myfinger[loopbit] );
	myent->hashf = target_hash;
	myent->start = powow;
      }
    }
    return change;
  }
  
  void routedata (int route_from, int from_ip, int route_dest, int route_size, int route_transport, char *route_msg, int pathreq) {
    int should_forward, nextguy;
    neighbor_finger *predent;
    
    predent = neighbor_random(pred);
    
    if (isinrangeright(route_dest, predent->hashf, myhash)) { 
      // final destiation
    }
    else { // trying to route through
      nextguy = nexthop(route_dest);
      if (nextguy != me) { // I can route it
	if (!pathreq)
	  should_forward = upcall_forward(nextguy, route_msg, route_size, COMM_TYPE_UNICAST);
	else 
	  should_forward = 0;
	if (should_forward == 0) {
	  if (!pathreq) {  // no need to check cache, routing thru overlay
	    route_data( nextguy, route_dest, from_ip, route_from, route_transport, route_msg, route_size, route_transport);
	  }
	  else { // requesting cache services
	    int lower = map.query(route_dest);
	    if (lower) {   // he's in my cache
	      route_data( lower, route_dest, from_ip, route_from, route_transport, route_msg, route_size, route_transport);
	    }
	    else {
	      route_data_path_req( nextguy, route_dest, from_ip, route_from, route_transport, route_msg, route_size, route_transport);
	    }
	  }
	}
	return;
      }
    }

    // I must be the final destination
    // I may be the only node in the system
    should_forward = upcall_forward( 0, route_msg, route_size, COMM_TYPE_UNICAST);
    if (should_forward == 0) {
      //          printf("deliverance 1 dest %.8x, pred %.8x, mine %.8x\n", route_dest, predent->hashf, myhash);
      int low;
      if (predent->ipaddr == me)
	low = myhash;
      else
	low = predent->hashf-1;
      if(pathreq && from_ip!=me) { // send a mapping if possible
	route_mapping(from_ip, low, myhash, me, 0, 0, -1);
      }
      upcall_deliver( route_msg, route_size, COMM_TYPE_UNICAST);
    }
  }
}