pastry.mac

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

          route_inform_request(ipaddr,myhash,joined,0,0,-1);
        } else {
          if (sr) {
            add_right(ipaddr,id);
          }
          else  {
            insert_right(ipaddr,id);
            update_leafset(Lmax_ipaddr, Lmax, false, joined);
          }
        }
      }
      else {
        if ((sl || sr) && !knownLive) {
          route_inform_request(ipaddr,myhash,joined,0,0,-1);
        } else {
          if (sl && sr) {
            // use a heuristic to guess where it should go
            int d = arc_distance(myhash,id);
            if (d < 0) {	
              add_left(ipaddr,id);
            } else {
              add_right(ipaddr,id);
            }  
          }
          else if (sl)    
            add_left(ipaddr,id);
          else if (sr)    
            add_right(ipaddr,id);
        }
      }
    }
  }

  int make_routing_decision(int id, int& rare_case)
  {

    int dist, d, closest = INT_MAX;
    neighbor_leafset* closestleaf = NULL;

    rare_case = 0;

    if(isinrange(id,start_space,end_space)) { //if I manage it, done.
      return me;
    }
    else if (neighbor_empty(myleft) && isinrange(id,Lmax+1,Lmin)) {
      //I don't manage it, but it's between Lmax and me.  Therefore, Lmax manages it.
      int nexthop;
      int nexthopid;

      nexthop=neighbor_info(myhashleafset,Lmax,id);
      nexthopid=Lmax;

      //printf("DEBUG: (empty left leafset) me:%.8x(%.8x), id:%.8x, nh:%.8x(%.8x), Lmax:%.8x, Lmin:%.8x\n",myhash,me,id,nexthopid,nexthop,Lmax,Lmin);
      //fflush(stdout);
      return nexthop;  
    }
    else if (neighbor_empty(myright) && isinrange(id,Lmax+1,Lmin)) {
      //I don't manage it, but it's between Lmin and me.  Therefore, Lmin manages it.
      int nexthop;
      int nexthopid;
      //printf("DEBUG: abs(arc_distance(id, myhash)) [%d] < abs(arc_distance(id, Lmin)) [%d] = [%d]\n",abs(arc_distance(id, myhash)),abs(arc_distance(id, Lmin)),(abs(arc_distance(id, myhash)) < abs(arc_distance(id, Lmin))));
      //fflush(stdout);
      nexthop=neighbor_info(myhashleafset,Lmin,id);
      nexthopid=Lmin;

      //printf("DEBUG: (empty right leafset) me:%.8x(%.8x), id:%.8x, nh:%.8x(%.8x), Lmax:%.8x, Lmin:%.8x\n",myhash,me,id,nexthopid,nexthop,Lmax,Lmin);
      //fflush(stdout);
      return nexthop;  
    }
    else if (!neighbor_empty(myleafset) && isinrange(id,Lmin,Lmax+1)) {
      // id is within range of our leaf set (and I don't manage it)
      foreach_neighbor(neighbor_leafset*, leaf, myleafset) {
        dist = abs(arc_distance(id,leaf->id));
        if (dist < closest) {
          closest = dist;
          closestleaf = leaf;
        }
      }
      int nexthop;
      int nexthopid;
      nexthop= closestleaf->ipaddr;
      nexthopid= closestleaf->id;

      //printf("DEBUG: (using leafset) me:%.8x(%.8x), id:%.8x, nh:%.8x(%.8x), Lmax:%.8x, Lmin:%.8x\n",myhash,me,id,nexthopid,nexthop,Lmax,Lmin);
      //fflush(stdout);
      return nexthop;
    }
    else {
      // use the routing table
      int l = shared_prefix_length(id, myhash);
      int Dl = nth_digit(id,l);

      if (!neighbor_empty(mytable[l][Dl])) {
        //printf("DEBUG: (using routing table) me:%.8x, id:%.8x, nh:%.8x\n",myhash,id,neighbor_random(mytable[l][Dl])->ipaddr);
        //fflush(stdout);
        return neighbor_closest(mytable[l][Dl])->ipaddr;
      }
      else {
        // "rare case"
        rare_case = 1;

        // return the closest (to 'id') entry in our state
        // tables that has a shared prefix at least as long as 'l' 

        int best = abs(arc_distance(id,myhash));
        int best_hop = me;

        foreach_neighbor(neighbor_leafset*,entry,myleafset) {
          int l2 = shared_prefix_length(id,entry->id);
          int d2 = abs(arc_distance(id,entry->id));
          if (l2 >= l && d2 < best) {
            best = d2;
            best_hop = entry->ipaddr;
          }
        }

        // rows 0 through (l-1) won't have a long enough prefix
        for (int r=l; r<ROWS; r++)
          for (int c=0; c<COLS; c++) {
            // i.e. if there really is an entry
            if (!neighbor_empty(mytable[r][c])) {
              neighbor_routeset *entry = neighbor_random(mytable[r][c]);
              int l2 = shared_prefix_length(id,entry->id);
              int d2 = abs(arc_distance(id,entry->id));
              if (l2 >= l && d2 < best) {
                best = d2;
                best_hop = entry->ipaddr;
              }
            }
          }

        //printf("DEBUG: (using routing table rare) me:%.8x, id:%.8x, nh:%.8x\n",myhash,id,best_hop);
        //fflush(stdout);
        return best_hop;
      }
    }
  }

  // Called when we have received all initial state 
  // to make sure that joining information we gathered is not stale
  void informAll()
  {
    // send info_reply msgs to see if the state sent
    // before is now stale

    route_leafset_info_reply(leafset_from, myhash, leafset_time_sent, 0, 0, -1);
    for (int i=0; i<ROWS; i++) {
      route_row_info_reply(row_from[i], myhash, row_time_sent[i], i, 0, 0, -1);
    }

    // inform my neighbors and send them my state
    foreach_neighbor(neighbor_leafset*,n,myleafset) {
      route_inform(n->ipaddr,myhash, 0, 0, -1);
    }

    for (int r=0; r<ROWS; r++) {
      for (int c=0; c<COLS; c++) { 
        foreach_neighbor(neighbor_routeset*,n,mytable[r][c]) {
          if (n->ipaddr != me) {
            route_row_info(n->ipaddr,myhash,curtime,r,mytable[r], 0, 0, -1);
          }
        }
      }
    }  

  }

  // Sends the row r to the node addr
  void sendrow(int addr, int r) 
  {
    route_row_info(addr,myhash,curtime,r,mytable[r], 0, 0, -1);
  }

  // Determine the next hop for the destination, id
  // If it is me, deliver it to me, otherwise propagate
  // Note that rare_case may be when making the route decision.
  // This routine also sets the nexthop
  int routedata(int id, char *msg, int size, int transport, int& rare_case, int& nexthop)
  {
    rare_case = 0;
    nexthop = make_routing_decision(id, rare_case);

    if( upcall_forward(nexthop, msg, size, COMM_TYPE_UNICAST) == 0 )
    {
      if (nexthop == me) {
        upcall_deliver(msg, size, COMM_TYPE_UNICAST);
        return 0;
      }
      else {
        route_data(nexthop, id, rare_case, myhash, transport, msg, size, transport);
        return macedon_sendret;
      }
    }
    return 0; //returning 0 because upper layer told me to drop it -- hence there were no problems sending it.
  }

  int routedatapathreq(int dest, int source_ip, int source_hash, char *msg, int size, int transport, int iamsender)
  {
    int rare_case;
    int nexthop = make_routing_decision(dest, rare_case);

    if( (!iamsender && nexthop != me) ||
        ( (iamsender || nexthop == me) &&
          upcall_forward(nexthop, msg, size, COMM_TYPE_UNICAST) == 0) )
    {
      if (nexthop == me) {
        if(source_ip != me) {
          route_mapping(source_ip, start_space, end_space, me, 0, 0, -1);
        } else {
          map.insert(start_space, end_space, me);  
        }
        upcall_deliver(msg, size, COMM_TYPE_UNICAST);
        return 0;
      }
      else {
        route_data_path_req(nexthop, dest, source_ip, source_hash, myhash, transport, msg, size, transport);
        return macedon_sendret;
      }
    }
    return 0; //I am an edge, and was told not to forward this.  "No problems"
  }

  void checkFinishedJoining()
  {

    for (int i=ROWS-1; i>=0; i--) {
#ifdef JOINING_TRACE
      if(row_received[i] == 0) {
        debug_macro("Pastry: checkFinishedJoining() not finished, need row %d.\n",i);
      }
#endif
      if (row_received[i] == 0) { 
        return; 
      }
    }

    if (leafset_received) {
#ifdef JOINING_TRACE
      debug_macro("Pastry: checkFinishedJoining() finished, informing all.\n");
#endif
      timer_cancel(join_timer);
      state_change(joined);
      informAll();
    }
#ifdef JOINING_TRACE
    else {
      debug_macro("Pastry: checkFinishedJoining() not finished, need leafset.\n");
    }
#endif
  }

  bool range_left(int id)
  {
    return neighbor_empty(myleft) ? false : isinrange(id,Lmin+1,myhash);
  }

  bool range_right(int id)
  {
    return neighbor_empty(myright) ? false : isinrange(id,myhash,Lmax);
  }

  void add_left(int addr, int id)
  {
    //debug_macro("DEBUG: Adding node %.8x(%.8x) to leafset (left)\n",id,addr);

    neighbor_add(myleafset,addr);
    neighbor_info(myleafset,addr,id) = id;

    neighbor_add(myhashleafset,id);
    neighbor_info(myhashleafset,id,id) = addr;

    neighbor_add(myleft,addr);
    neighbor_info(myleft,addr,id) = id;

    recompute_leaf_bounds();
    upcall_notify(myhashleafset, NBR_TYPE_PEER);
  }

  void add_right(int addr, int id)
  {
    //debug_macro("DEBUG: Adding node %.8x(%.8x) to leafset (right)\n",id,addr);

    neighbor_add(myleafset,addr);
    neighbor_info(myleafset,addr,id) = id;

    neighbor_add(myhashleafset,id);
    neighbor_info(myhashleafset,id,id) = addr;

    neighbor_add(myright,addr);
    neighbor_info(myright,addr,id) = id;		

    recompute_leaf_bounds();
    upcall_notify(myhashleafset, NBR_TYPE_PEER);
  }

  void insert_left(int addr, int id)
  {
    //debug_macro("DEBUG: Adding node %.8x(%.8x) to leafset (left)\n",id,addr);

    neighbor_remove(myhashleafset,neighbor_info(myleafset,Lmin_ipaddr,id));

    neighbor_remove(myleafset,Lmin_ipaddr);
    neighbor_add(myleafset,addr);
    neighbor_info(myleafset,addr,id) = id;
    neighbor_add(myhashleafset,id);
    neighbor_info(myhashleafset,id,id) = addr;

    neighbor_remove(myleft,Lmin_ipaddr);
    neighbor_add(myleft,addr);
    neighbor_info(myleft,addr,id) = id;

    recompute_leaf_bounds();
    upcall_notify(myhashleafset, NBR_TYPE_PEER);
  }

  void insert_right(int addr, int id)
  {
    //debug_macro("DEBUG: Adding node %.8x(%.8x) to leafset (right)\n",id,addr);

    neighbor_remove(myhashleafset,neighbor_info(myleafset,Lmax_ipaddr,id));

    neighbor_remove(myleafset,Lmax_ipaddr);
    neighbor_add(myleafset,addr);
    neighbor_info(myleafset,addr,id) = id;

    neighbor_add(myhashleafset,id);
    neighbor_info(myhashleafset,id,id) = addr;

    neighbor_remove(myright,Lmax_ipaddr);
    neighbor_add(myright,addr);
    neighbor_info(myright,addr,id) = id;

    recompute_leaf_bounds();
    upcall_notify(myhashleafset, NBR_TYPE_PEER);
  }	

  // Determines the farthest leaf on the each side of leafset 
  void recompute_leaf_bounds()
  {
    unsigned int lowest, highest, d;
    leafset_timestamp = curtime;

    lowest = UINT_MAX;
    highest = 0;
    foreach_neighbor(neighbor_halfset*,leaf,myleft) {
      d = uarc_distance(myhash,leaf->id);
      if (d < lowest) {
        lowest = d;
        Lmin = leaf->id;
        Lmin_ipaddr = leaf->ipaddr;
      }
      if (d > highest) {
        highest = d;
        start_space = myhash - (UINT_MAX-highest+2)/2; //+1 was missing, an error on UINT_MAX(+1), another +1 was missing as to break ties.
      }
    }

    lowest = UINT_MAX;
    highest = 0;
    foreach_neighbor(neighbor_halfset*,leaf,myright) {
      d = uarc_distance(myhash,leaf->id);			
      if (d > highest) {
        highest = d;
        Lmax = leaf->id;
        Lmax_ipaddr = leaf->ipaddr;
      }
      if (d < lowest) {
        lowest=d;
        end_space = myhash + (lowest/2);
      }
    }

    if(neighbor_empty(myleft) && neighbor_empty(myright)) {
      start_space = end_space = myhash;
    } else 
    if(neighbor_empty(myleft)) {
      start_space = myhash - (uarc_distance(Lmax, myhash)+1)/2;
    } else 
    if(neighbor_empty(myright)) {
      end_space = myhash + (uarc_distance(myhash, Lmin))/2;
    }
    map.insert(start_space, end_space, me);  
  }

  // returns the length of the shorter arc between two points
  // on the circle (i.e. between the two ids in id-space).
  // positive values: 'to' comes after 'from'
  // negative values: 'to' comes before 'from'
  // in absolute terms, this distance can be as large as half the circle
  int arc_distance(unsigned int from, unsigned int to)
  {
    if (from==to) // Same point, zero distance
      return 0;
    unsigned int d;

    if (to > from) {
      d = to - from;
      if (d > (unsigned int)INT_MAX)
        return -(int)(UINT_MAX - d +1);
      else
        return (int) d;
    }
    else {
      d = from - to;
      if (d > (unsigned int)INT_MAX)
        return (int)(UINT_MAX - d +1);
      else
        return -(int) d;
    }
  }

  // returns the arc length from
  // 'from' to 'to'; always a nonnegative number 
  // this distance may go almost the entire way around the circle
  unsigned int uarc_distance(unsigned int from, unsigned int to)
  {
    int d = arc_distance(from,to);
    if (d >= 0)
      return d;
    else
      return UINT_MAX - (unsigned int)(-d) + 1;
  }
}