fm.c

基于linux环境的ns2多机并行仿真补丁

			if( i < FM_numnodes )
			{
			    int j = grp->numnodes++;
			    MYASSERT( j < FMMAXPE,
			      ("Too many nodes %d in subnet %s",j,nettype_str));
			    strcpy( grp->canname[j], othername );
			    grp->fmnodeid[j] = i;
			}
		    }
		}

		if( grp->numnodes > 1 )
		{
if(mixfm->dbg>=1){printf("Note: %d nodes used from subnet %s on line %d.\n",grp->numnodes,nettype_str,linenum);}
		}
		else
		{
		    --grps->ngroups;
if(mixfm->dbg>=1){printf("Note: No links belonging to subnet %s on line %d are\nused. Here is the corresponding subnet specification:\n\t\"%s\"\n",nettype_str,linenum,line);}
		}
	    }
	}
        fclose(fp);
    }
}

/*---------------------------------------------------------------------------*/
static void group_nodes( void )
{
    #if MPI_AVAILABLE
    group_mpi_nodes();
    #else /*MPI_AVAILABLE*/
    group_shm_nodes();
    #endif /*MPI_AVAILABLE*/
    read_network_file();
}

/*---------------------------------------------------------------------------*/
static void print_adj_matrix( void )
{
    AdjMatrix *adj = &mixfm->net.adj;
    int i = 0, j = 0;

	printf("-----------------------------------\n");
	printf("%10.10s%3.3s", "", "");
        for( i = 0; i < FM_numnodes; i++ )
        {
            printf("%5.5s%-4d%5.5s", "", i, "");
        }
	printf("\n");
        for( i = 0; i < FM_numnodes; i++ )
        {
	    printf("%10.10s%3d", mixfm->FM_nodenames[i], i);
            for( j = 0; j < FM_numnodes; j++ )
            {
	        AdjEntry *aij = &adj->matrix[i][j];
		if( aij->xf_cltag == FM_XFACE_CLASS_NUL )
		{
		    printf("%14.14s","");
		}
		else
		{
	            printf("%3.3s%3.3s%-3d->%-3d",
		           "", XFACE_CLASS_STR(aij->xf_cltag),
		           aij->subnet_id, aij->next_hop);
		}
            }
	    printf("\n");
        }
	printf("-----------------------------------\n");
}

/*---------------------------------------------------------------------------*/
static void make_adj_matrix( void )
{
    int i = 0, j = 0, k = 0, x = 0, y = 0, g = 0, s = 0;
    AdjMatrix *adj = &mixfm->net.adj;

    /*Initialize*/
    for( i = 0; i < FM_numnodes; i++ )
    {
	for( j = 0; j < FM_numnodes; j++ )
	{
            AdjEntry *ae = &adj->matrix[i][j];
	    ae->xf_cltag = FM_XFACE_CLASS_NUL;
	    ae->subnet_id = -1;
	    ae->next_hop = -1;
	    ae->cost = LINK_COST_INF;
	}
    }

    /*Set up subnets info*/
    for( s = 0; s < FM_SUBNET_NUM; s++ )
    {
      NodeGroupMap *grps = &mixfm->net.grps[s];
      for( g = 0; g < grps->ngroups; g++ )
      {
	NodeGroup *grp = &grps->group[g];
	if( grp->numnodes > 1 )
	{
	  int subnet_id = mixfm->net.tot_subnets++;
	  mixfm->net.subnets[subnet_id] = grp;
if(mixfm->dbg>=2){printf("Subnet type %d group %d ID %d\n",s,g,subnet_id);}
	  for( x = 0; x < grp->numnodes; x++ )
	  {
	    for( y = 0; y < grp->numnodes; y++ )
	    {
	      if( x == y )
	      {
	      }
	      else
	      {
		  i = grp->fmnodeid[x];
		  j = grp->fmnodeid[y];
		  MYASSERT( 0 <= i < FM_numnodes, ("%d",i) );
		  MYASSERT( 0 <= j < FM_numnodes, ("%d",j) );
		  {
		      AdjEntry *ae = &adj->matrix[i][j];
		      if( SUBNET_LINK_COST(s) < ae->cost )
		      {
		          ae->xf_cltag = SUBNET_TYPE_TO_XFACE_CLASS(s);
		          ae->subnet_id = subnet_id;
		          ae->next_hop = j;
		          ae->cost = SUBNET_LINK_COST(s);
		      }
		  }
	      }
	    }
	  }
	}
      }
    }

if(mixfm->dbg>=3){printf("Direct links:\n");print_adj_matrix();}

    /*Compute nexthops of shortest-paths, via transitive closure*/
    for( k = 0; k < FM_numnodes; k++ )
    {
        for( i = 0; i < FM_numnodes; i++ )
	{
            for( j = 0; j < FM_numnodes; j++ )
	    {
	        AdjEntry *aik = &adj->matrix[i][k];
		AdjEntry *akj = &adj->matrix[k][j];
		AdjEntry *aij = &adj->matrix[i][j];

		if( i == j ) continue; /*Skip loopbacks*/

		if(
		    /*No direct route i->j, but can go i->k->j*/
		    ( aij->xf_cltag == FM_XFACE_CLASS_NUL &&
		      aik->xf_cltag != FM_XFACE_CLASS_NUL &&
		      akj->xf_cltag != FM_XFACE_CLASS_NUL )
		    ||
		    /*Route i->k->j is cheaper than current route i->j*/
		    ( aij->xf_cltag != FM_XFACE_CLASS_NUL &&
		      aik->xf_cltag != FM_XFACE_CLASS_NUL &&
		      akj->xf_cltag != FM_XFACE_CLASS_NUL &&
		      aik->cost+akj->cost < aij->cost )
		  )
		  /*Route from i to j using the i->k direct link*/
		{
		    aij->xf_cltag = aik->xf_cltag;
		    aij->subnet_id = aik->subnet_id;
		    aij->next_hop = aik->next_hop;
		    aij->cost = aik->cost + akj->cost;
		}
		else
		{
		    /*Do nothing*/
		}
	    }
	}
    }

if(mixfm->dbg>=2){printf("Next hops transitive closure:\n");print_adj_matrix();}

    for( i = 0; i < FM_numnodes; i++ )
    {
        for( j = 0; j < FM_numnodes; j++ )
	{
	    if( i != j )
	    {
	        AdjEntry *aij = &adj->matrix[i][j];
	        MYASSERT( aij->xf_cltag != FM_XFACE_CLASS_NUL,
		        ("\n\nNode #%d %s unreachable from node #%d %s.\n"
			 "Check your network information file.",
			 j, mixfm->FM_nodenames[j], i, mixfm->FM_nodenames[i]));
	    }
	}
    }
}

/*---------------------------------------------------------------------------*/
static void instantiate_interfaces( void )
{
    int subnet_id = 0;
    AdjMatrix *adj = &mixfm->net.adj;

    mixfm->nxfaces = 0;
    for( subnet_id = 0; subnet_id < mixfm->net.tot_subnets; subnet_id++ )
    {
	char incl[FMMAXPE];
        XFaceClassTag cltag = FM_XFACE_CLASS_BAD;
	int i = 0, j = 0, nincl = 0;
	for( i = 0; i < FM_numnodes; i++ ) incl[i] = 0;
	for( i = 0; i < FM_numnodes; i++ )
	{
	    for( j = 0; j < FM_numnodes; j++ )
	    {
	        AdjEntry *aij = &adj->matrix[i][j];
		if( i != j && aij->xf_cltag != FM_XFACE_CLASS_NUL &&
		    aij->subnet_id==subnet_id && aij->next_hop==j )
		{
		    if( !incl[i] ) { incl[i] = 1; nincl++; }
		    if( !incl[j] ) { incl[j] = 1; nincl++; }
		    if( cltag == FM_XFACE_CLASS_BAD ) cltag = aij->xf_cltag;
		    MYASSERT( cltag==aij->xf_cltag,("%d %d",cltag,aij->xf_cltag));
		}
	    }
	}

	MYASSERT( use_shm || !(cltag == FM_XFACE_CLASS_SHM && incl[FM_nodeid]),
	        ("Can't use SHM on this federate.\n") );
	MYASSERT( use_gm || !(cltag == FM_XFACE_CLASS_MYR && incl[FM_nodeid]),
	        ("Can't use GM/Myrinet on this federate.\n") );
	MYASSERT( use_tcp || !(cltag == FM_XFACE_CLASS_TCP && incl[FM_nodeid]),
	        ("Can't use TCP on this federate.\n") );
	MYASSERT( use_mpi || !(cltag == FM_XFACE_CLASS_MPI && incl[FM_nodeid]),
	        ("Can't use MPI on this federate.\n") );

	if( incl[FM_nodeid] ) /*I am in this subnet*/
	{
	    /*Need to instantiate an interface for this subnet*/

	    char *snm = (char *)malloc( 10*sizeof(char) );
	    int numnodes = 0, myid = -1;
	    NodeGroup *sgrp = mixfm->net.subnets[subnet_id];
	    XFace *xf = &mixfm->xfaces[mixfm->nxfaces++];

	    xf->xf_class = &xf_classes[cltag];
	    sprintf(snm,"%s%d",XFACE_CLASS_STR(cltag),subnet_id);
	    xf->xf_name = snm;
	    xf->xf_subnet_id = subnet_id;
	    xf->xf_constructed = 0;

	    /*For this subnet, find #nodes, map, inv_map, cannames & ID*/
	    for( i = 0; i < FM_numnodes; i++ )
	    {
		if( !incl[i] )
		{
		    xf->xf_map[i] = -1;
		}
		else
		{
		    int id = numnodes++, m = 0;
		    if( i == FM_nodeid ) myid = id;
		    xf->xf_map[i] = id;
		    xf->xf_inv_map[id] = i;

		    for( m = 0; m < sgrp->numnodes; m++ )
		    {
		        if( sgrp->fmnodeid[m] == i ) break;
		    }
		    MYASSERT( m < sgrp->numnodes, ("%d %d",subnet_id,i) );
		    strcpy( xf->xf_canname[id], sgrp->canname[m] );
		}
	    }

	    MYASSERT( nincl == numnodes, ("%d %d",nincl,numnodes) );
	    MYASSERT( 2 <= numnodes && numnodes <= FM_numnodes,
	            ("%d %ld",numnodes,FM_numnodes) );
	    MYASSERT( 0 <= myid && myid < numnodes, ("%d %d",myid,numnodes) );

	    xf->xf_sub_id = myid;
	    xf->xf_sub_n = numnodes;

	    xf->xf_class->xf_new( xf );

if(mixfm->dbg>=2){print_xface( xf );}
        }
    }
}

/*---------------------------------------------------------------------------*/
static void make_nexthop_table( void )
{
    int i = 0, j = 0;
    AdjMatrix *adj = &mixfm->net.adj;

    for( i = FM_nodeid, j = 0; j < FM_numnodes; j++ )
    {
	AdjEntry *aij = &adj->matrix[i][j];
        if( i == j )
	{
	}
	else
	{
	    int x = 0;
	    XFace *found_xf = 0;
	    for( x = 0; x < mixfm->nxfaces; x++ )
	    {
	        XFace *xf = &mixfm->xfaces[x];
	        if( xf->xf_subnet_id == aij->subnet_id )
		{
		    found_xf = xf;
		    break;
		}
	    }
	    MYASSERT( found_xf, ("Interface not found for dest FM node %d", j) );
	    {
	        XFaceNextHopEntry *nhe =&mixfm->net.nhop_table.nhop_entry[j];
	        nhe->dest_id = j;
	        nhe->next_hop = aij->next_hop;
	        nhe->xf = found_xf;
	    }
	}
    }
}

/*---------------------------------------------------------------------------*/
static void initialize_interfaces( void )
{
    int s = 0, x = 0;

    /*Initialize the interfaces in ascending order of subnet type*/
    for( s = 0; s < FM_SUBNET_NUM; s++ )
    {
        for( x = 0; x < mixfm->nxfaces; x++ )
	{
	    XFace *xf = &mixfm->xfaces[x];
	    if( !xf->xf_constructed &&
	        xf->xf_class->xf_cltag == SUBNET_TYPE_TO_XFACE_CLASS(s) )
	    {
		int myid = xf->xf_sub_id, numgrpnodes = xf->xf_sub_n;
		FMNodeName *cannames = xf->xf_canname;
	        xf->xf_class->xf_constructor( xf, myid, numgrpnodes, cannames );
		xf->xf_constructed = 1;
	    }
	}
    }
}

/*---------------------------------------------------------------------------*/
static void finalize_interfaces( void )
{
    int s = 0, x = 0;

    /*Finalize the interfaces in descending order of subnet type*/
    for( s = FM_SUBNET_NUM-1; s >= 0; --s )
    {
        for( x = mixfm->nxfaces-1; x >= 0; --x )
	{
	    XFace *xf = &mixfm->xfaces[x];
	    if( xf->xf_class->xf_cltag == SUBNET_TYPE_TO_XFACE_CLASS(s) )
	    {
	        xf->xf_class->xf_destructor( xf );
	    }
	}
    }
}

/*---------------------------------------------------------------------------*/
/*                                                                           */
/*---------------------------------------------------------------------------*/
void FM_pre_init( int *pac, char ***av ) /*Appl MUST call PreInit if using MPI*/
{
    #if MPI_AVAILABLE
	FMMPI_pre_initialize( pac, av );
	/*Create NODEINFO env var*/
	{
	    char *hname = "localhost";
	    char *nstr = (char *)malloc(FMMAXHOSTNAMELEN+200);
	    sprintf(nstr,"NODEINFO=%d:%d@%s:%d",
		    FMMPI_numnodes,FMMPI_numnodes,hname,FMMPI_nodeid);
	    putenv(nstr);
	}
    #endif /*MPI_AVAILABLE*/
}

/*---------------------------------------------------------------------------*/
void FM_initialize( void )
{
    FM_handler_table=(FM_handler**)calloc(FM_MAX_HANDLERS, sizeof(FM_handler*));
    MYASSERT( FM_handler_table, ("!") );
    mixfm = (MixFMData *)calloc( 1, sizeof( MixFMData ) );
    MYASSERT(mixfm,("Can't allocate %d bytes for MixFMData",sizeof(MixFMData)));

    {char *estr = getenv("FM_DEBUG"); mixfm->dbg = estr ? atoi(estr) : 0;}

if(mixfm->dbg>=1){printf("FM_DEBUG=%d\n",mixfm->dbg);fflush(stdout);}
if(mixfm->dbg>=0){printf("FM_initialize() started.\n");fflush(stdout);}
if(mixfm->dbg>=0){printf("Sizeof(MixFMData)== %.2lf MB.\n",sizeof(MixFMData)/1e6);}

    config();
    group_nodes();
    make_adj_matrix();
    instantiate_interfaces();
    make_nexthop_table();
    initialize_interfaces();

if(mixfm->dbg>=0){printf("FM_initialize() done.\n");fflush(stdout);}
}

/*---------------------------------------------------------------------------*/
void FM_finalize( void )
{
    finalize_interfaces();
}

/*---------------------------------------------------------------------------*/
FM_stream *FM_begin_message( ULONG to_fm_id, ULONG length, ULONG handler )
{
    MYASSERT( to_fm_id < FM_numnodes, ("%ld",to_fm_id) );
    MYASSERT( to_fm_id != FM_nodeid, ("Can't send to self %ld", to_fm_id) );

    {
      XFaceNextHopEntry *nhe =&mixfm->net.nhop_table.nhop_entry[to_fm_id];
      int nh = nhe->next_hop;
      XFace *xf = nhe->xf;

if(mixfm->dbg>=3){printf("%s from %ld via %d to %ld\n", (nh==to_fm_id?"Directly sending":"Indirectly forwarding"), FM_nodeid, nh, to_fm_id);}

      MYASSERT( nhe->dest_id == to_fm_id, ("%d %ld",nhe->dest_id,to_fm_id) );
      MYASSERT( 0 <= nh && nh < FM_numnodes, ("%d",nh) );
      MYASSERT( xf, ("Interface must exist") );
      {
        int m = xf->xf_map[nh];
	MYASSERT( 0 <= m && m < xf->xf_sub_n, ("%d %d",m,xf->xf_sub_n) );
	MYASSERT( xf->xf_inv_map[m] == nh,
	        ("Circular check %d %d",xf->xf_inv_map[m], nh) );
	MYASSERT( m != xf->xf_sub_id, ("Can't send to self") );
	mixfm->curr.out.xf = xf;