p4_md.c

MPICH是MPI的重要研究,提供了一系列的接口函数,为并行计算的实现提供了编程环境.

        }
	p4_dprintfl(10, "sent msg of type %d from %d to %d via NCUBE send\n",
		    m->type,m->from,m->to);
    }
    else
    {
	/* Send a message, asking for an acknowledgement of receipt. */
	rc = nwrite(m, len, to, ACK_REQUEST_NCUBE, &unused_flag);
        if (rc < 0)
        {
            p4_dprintf("nwrite failed for msg of length %d from %d to %d \n",
                        len,m->from,m->to);
            p4_error("exiting due to failed nwrite",rc);
        }
	/* Wait for the acknowledgement. */
	p4_dprintfl(30, "waiting for ack from %d\n", m->to);
        type = ACK_REPLY_NCUBE;
        nread(&buf, sizeof(char), &to,  &type, &unused_flag);
	p4_dprintfl(30, "received ack from %d\n", m->to);
	p4_dprintfl(10, "sent msg of type %d from %d to %d via NCUBE csend\n",m->type,m->from,m->to);
    }
}

P4BOOL MD_NCUBE_msgs_available()
{
    P4BOOL rc;
    int from, type;

    from = NCUBE_ANY_NODE;
    type = NCUBE_ANY_TAG;
    rc = ntest(&from, &type);
    if (rc == -1)
        rc = 0;
    return (rc);
}

/* endif for ncube */
#endif


#if defined(IPSC860)  ||  defined(CM5)  || defined(NCUBE) \
                      ||  defined(SP1_EUI) || defined(SP1_EUIH)

int ns_start(argc, argv)
int *argc;
char **argv;
{
    char *s;
    char ns_host[100];
    struct bm_rm_msg bm_msg;
    int from, type, len, to, unused_flag;

    sprintf(whoami_p4, "ns_%d_%d", MYNODE(), getpid());

#   if defined(IPSC860)
    crecv(SYNC_MSG, &bm_msg, (long) sizeof(struct bm_rm_msg));
#   endif 
#   if defined(CM5)
    CMMD_receive(CMMD_ANY_NODE, CMMD_ANY_TAG, (void *) &bm_msg, sizeof(struct bm_rm_msg));
    if (CMMD_msg_tag() != SYNC_MSG) /* should be a DIE, message, otherwise */
        exit(0);
#   endif 
#   if defined(NCUBE)
    from = NCUBE_ANY_NODE;
    type = NCUBE_ANY_TAG;
    nread(&bm_msg, sizeof(struct bm_rm_msg), &from,  &type, &unused_flag);
    if (type != SYNC_MSG)  /* should be a DIE, message, otherwise */
        exit(0);
#   endif 
#   if defined(SP1_EUI)
    from = ANY_P4TYPE_EUI;
    type = ANY_P4TYPE_EUI;
    mpc_brecv(&bm_msg, sizeof(struct bm_rm_msg), &from,  &type, &unused_flag);
    if (type != SYNC_MSG)  /* should be a DIE, message, otherwise */
        exit(0);
#   endif 
#   if defined(SP1_EUIH)
    from = ANY_P4TYPE_EUIH;
    type = ANY_P4TYPE_EUIH;
    len  = sizeof(struct bm_rm_msg);
    mp_brecv(&bm_msg, &len, &from,  &type, &unused_flag);
    if (type != SYNC_MSG)  /* should be a DIE, message, otherwise */
        exit(0);
#   endif 

    /* Send off my info to my rm for forwarding to bm */
    bm_msg.type = p4_i_to_n(REMOTE_SLAVE_INFO);
    bm_msg.slave_idx = p4_i_to_n(MYNODE());
    bm_msg.slave_pid = p4_i_to_n(getpid());
    bm_msg.switch_port = p4_i_to_n(-1);
    ns_host[0] = '\0';
    get_qualified_hostname(ns_host,100);
    strcpy(bm_msg.host_name,ns_host);

#   if defined(IPSC860)
    csend((long) INITIAL_INFO, &bm_msg, (long) sizeof(struct bm_rm_msg), (long) 0, (long) NODE_PID);
    crecv(INITIAL_INFO, &bm_msg, (long) sizeof(struct bm_rm_msg));
#   endif 
#   if defined(CM5)
    CMMD_send_noblock(0, INITIAL_INFO, &bm_msg, sizeof(struct bm_rm_msg));
    CMMD_receive(CMMD_ANY_NODE, INITIAL_INFO, (void *) &bm_msg, sizeof(struct bm_rm_msg));
#   endif 
#   if defined(NCUBE)
    nwrite(&bm_msg, sizeof(struct bm_rm_msg), 0, INITIAL_INFO, &unused_flag);
    from = NCUBE_ANY_NODE;
    type = NCUBE_ANY_TAG;
    nread(&bm_msg, sizeof(struct bm_rm_msg), &from,  &type, &unused_flag);
#   endif 
#   if defined(SP1_EUI)
    mpc_bsend(&bm_msg, sizeof(struct bm_rm_msg), 0, INITIAL_INFO);
    from = ANY_P4TYPE_EUI;
    type = ANY_P4TYPE_EUI;
    mpc_brecv(&bm_msg, sizeof(struct bm_rm_msg), &from,  &type, &unused_flag);
#   endif 
#   if defined(SP1_EUIH)
    type = INITIAL_INFO;
    to   = 0;
    len  = sizeof(struct bm_rm_msg);
    mp_bsend(&bm_msg, &len, &to, &type);
    from = ANY_P4TYPE_EUIH;
    type = ANY_P4TYPE_EUIH;
    len  = sizeof(struct bm_rm_msg);
    mp_brecv(&bm_msg, &len, &from,  &type, &unused_flag);
#   endif 

    if (strcmp(bm_msg.version,P4_PATCHLEVEL) != 0)
    {
	p4_dprintf("my version is %s\n",P4_PATCHLEVEL);
	p4_error("version does not match master",0);
    }
    if ((s = (char *) rindex(bm_msg.pgm,'/'))  !=  NULL)
    {
	*s = '\0';  /* chg to directory name only */
	chdir(bm_msg.pgm);
    }
    globmemsize = p4_n_to_i(bm_msg.memsize);
    logging_flag = p4_n_to_i(bm_msg.logging_flag);
    if (logging_flag)
	ALOG_ENABLE;
    else
	ALOG_DISABLE;

    MD_initmem(globmemsize);
    alloc_global();  /* sets p4_global */
    p4_local = alloc_local_rm();
    p4_global->num_in_proctable = p4_n_to_i(bm_msg.numinproctab);
    p4_global->local_slave_count = p4_n_to_i(bm_msg.numslaves);
    p4_debug_level = p4_n_to_i(bm_msg.debug_level);
    strcpy(p4_global->application_id, bm_msg.application_id);

#   if defined(IPSC860)
    crecv(INITIAL_INFO, p4_global->proctable, (long) sizeof(p4_global->proctable));
#   endif 
#   if defined(CM5)
    CMMD_receive(CMMD_ANY_NODE, INITIAL_INFO, (void *) p4_global->proctable, sizeof(p4_global->proctable));
#   endif 
#   if defined(NCUBE)
    from = NCUBE_ANY_NODE;
    type = INITIAL_INFO;
    nread((void *) p4_global->proctable, sizeof(p4_global->proctable), &from,  &type, &unused_flag);
#   endif 
#   if defined(SP1_EUI)
    from = ANY_P4TYPE_EUI;
    type = INITIAL_INFO;
    mpc_brecv(p4_global->proctable, sizeof(p4_global->proctable), &from,  &type, &unused_flag);
#   endif 
#   if defined(SP1_EUIH)
    from = ANY_P4TYPE_EUIH;
    type = INITIAL_INFO;
    len  = sizeof(p4_global->proctable);
    mp_brecv(p4_global->proctable, &len, &from, &type, &unused_flag);
#   endif 

    p4_local = alloc_local_slave();
    p4_local->listener_fd = -1;
    p4_local->my_id = p4_get_my_id_from_proc();
    sprintf(whoami_p4, "p%d_%d", p4_get_my_id(), getpid());

    setup_conntab();

    usc_init();
    init_usclock();
    ALOG_SETUP(p4_local->my_id,ALOG_TRUNCATE);
    ALOG_LOG(p4_local->my_id,BEGIN_USER,0,"");
    return(0);
}

/* endif for ifdef ipsc860 or cm5 */
#endif 

P4VOID MD_set_reference_time( void )
{
/* We want MD_clock to deal with small numbers */

#if defined(SYMMETRY_PTX)
/* reference time will be in seconds */
    struct timespec tp;
    getclock(TIMEOFDAY,&tp);
    p4_global->reference_time = tp.tv_sec;
#endif

#if defined(SUN)   || defined(RS6000)     || defined(DEC5000) \
 || defined(NEXT)  || defined(KSR)        || defined(CM5)     \
 || defined(SYMMETRY) || defined(BALANCE) || defined(LINUX)   \
 || defined(GP_1000)  || defined(TC_2000) || defined(CRAY)    \
 || defined(TITAN)    || defined(ALLIANT) || defined(SGI)     \
 || defined(NCUBE)    || defined(SP1_EUI) || defined(SP1_EUIH)\
 || defined(MULTIMAX) || defined(IBM3090) || defined(FREEBSD) \
 || defined(NETBSD) \
 || (defined(HP)  &&  !defined(SUN_SOLARIS))

/* reference time will be in seconds */
    struct timeval tp;
    struct timezone tzp;

    gettimeofday(&tp, &tzp);
    p4_global->reference_time = tp.tv_sec;
#endif

#if defined(SUN_SOLARIS)
#if defined(USE_WIERDGETTIMEOFDAY)
    struct timeval tp;

    gettimeofday(&tp);
#else
    struct timeval tp;

    gettimeofday(&tp,(void *)0);
#endif
    p4_global->reference_time = tp.tv_sec;
#endif

#if defined(IPSC860)  &&  !defined(MEIKO_CS2)
/* reference time will be in milliseconds */
    p4_global->reference_time = (unsigned long) (mclock());
#endif
    
}

int MD_clock( void )
{
    /* returns value in milleseconds */
    int i = 0;

#if defined(SYMMETRY_PTX)
    struct timespec tp;
    getclock(TIMEOFDAY,&tp);
    i = (int) (tp.tv_sec - p4_global->reference_time);
    i *= 1000;
    i += (int) (tp.tv_nsec / 1000000); /* On PTX the second field is nanosec */
#endif

#if defined(SUN)   || defined(RS6000)     || defined(DEC5000) \
 || defined(NEXT)  || defined(KSR)        || defined(CM5)     \
 || defined(SYMMETRY) || defined(BALANCE) || defined(LINUX)   \
 || defined(GP_1000)  || defined(TC_2000) || defined(CRAY)    \
 || defined(TITAN)    || defined(ALLIANT) || defined(SGI)     \
 || defined(NCUBE)    || defined(SP1_EUI) || defined(SP1_EUIH)\
 || defined(MULTIMAX) || defined(IBM3090) || defined(FREEBSD) \
 || defined(NETBSD) \
 || (defined(HP)  &&  !defined(SUN_SOLARIS))

    struct timeval tp;
    struct timezone tzp;

    gettimeofday(&tp, &tzp);
    i = (int) (tp.tv_sec - p4_global->reference_time);
    i *= 1000;
    i += (int) (tp.tv_usec / 1000);
#endif

#if defined(SUN_SOLARIS)
    struct timeval tp;
#if defined(USE_WIERDGETTIMEOFDAY)
    gettimeofday(&tp);
#else
    gettimeofday(&tp,(void *)0);
#endif
    i = (int) (tp.tv_sec - p4_global->reference_time);
    i *= 1000;
    i += (int) (tp.tv_usec / 1000);

#endif

#if defined(IPSC860)  &&  !defined(MEIKO_CS2)
    i = (int) (mclock() - p4_global->reference_time);
#endif

    return (i);
}

#ifdef DELTA
int myhost()
{
    return (0);
}
#endif


#ifdef SYSV_IPC

int init_sysv_semset(int setnum)
{
    int i, semid;
#   if defined(SUN_SOLARIS)
    union semun{
      int val;
      struct semid_ds *buf;
      ushort *array;
      } arg;
#   else
#    if defined(SEMCTL_ARG_UNION)
#       if !defined(SEMUN_UNDEFINED)
       union semun arg;
#       else
       union semun {
	   int val;
	   struct semid_ds *buf;
	   unsigned short int *array;
	   struct seminfo *__buf;
       } arg;
#       endif /* SEMUN_UNDEFINED */
#   else
#   if defined(IBM3090) || defined(RS6000) ||    \
       defined(TITAN)  || defined(DEC5000) ||    \
       defined(HP) || defined(KSR)  
    int arg;
#   else
#   if defined(SEMUN_UNDEFINED)    
       union semun {
	   int val;
	   struct semid_ds *buf;
	   unsigned short int *array;
	   struct seminfo *__buf;
       } arg;
#   else
       union semun arg;
#      endif
#   endif
#endif
#endif

#   if defined(SUN_SOLARIS) || defined(SEMCTL_ARG_UNION)
    arg.val = 1;
#   else
#   if defined(IBM3090) || defined(RS6000) ||    \
       defined(TITAN)  || defined(DEC5000) ||    \
       defined(HP) || defined(KSR) 
    arg = 1;
#   else
    arg.val = 1;
#   endif
#   endif

    if ((semid = semget(getpid()+setnum,10,IPC_CREAT|0600)) < 0)
    {
	p4_error("semget failed for setnum",setnum);
    }
    for (i=0; i < 10; i++)
    {
	if (semctl(semid,i,SETVAL,arg) == -1)
	{
	    p4_error("semctl setval failed",-1);
	}
    }
    return(semid);
}

P4VOID MD_lock_init(MD_lock_t *L)
{
int setnum;

    MD_lock(&(p4_global->slave_lock));
    setnum = p4_global->sysv_next_lock / 10;
    if (setnum > P4_MAX_SYSV_SEMIDS)
    {
	p4_error("exceeding max num of p4 semids",P4_MAX_SYSV_SEMIDS);
    }
    if (p4_global->sysv_next_lock % 10 == 0)
    {
	p4_global->sysv_semid[setnum] = init_sysv_semset(setnum);
	p4_global->sysv_num_semids++;
    }
    L->semid  = p4_global->sysv_semid[setnum];
    L->semnum = p4_global->sysv_next_lock - (setnum * 10);
    p4_global->sysv_next_lock++;
    MD_unlock(&(p4_global->slave_lock));
}


P4VOID MD_lock(MD_lock_t *L)
{
    sem_lock[0].sem_num = L->semnum;
    if (semop(L->semid,&sem_lock[0],1) < 0)
    {
	/* Use -1 so that we get the value from perror for the call */
	p4_error("OOPS: semop lock failed",-1); /* (int)L->semid); */
    }
}

P4VOID MD_unlock(MD_lock_t *L)
{
    sem_unlock[0].sem_num = L->semnum;
    if (semop(L->semid,&sem_unlock[0],1) < 0)
    {
        p4_error("OOPS: semop unlock failed",(int)L->semid);
    }
}
#endif
 

#if defined(SGI)  &&  defined(VENDOR_IPC)

/* MD_lock and MD_unlock are defined in p4_MD.h for SGI */

/* this is the spinlock method */
P4VOID MD_lock_init(MD_lock_t *L) 
{ 
    (*L) = usnewlock(p4_sgi_usptr);
}

/* this is the semaphore method */
/**********
P4VOID MD_lock_init(MD_lock_t *L) 
{ 
    (*L) = usnewsema(p4_sgi_usptr,1); 
}

**********/
#endif


#if defined(TCMP)

P4BOOL MD_tcmp_msgs_available(req_type,req_from)
int *req_from,*req_type;
{
    char *msg = NULL;
    int len_rcvd;
    tcmpfunp matcher;
    tcmp_status tcmpstat;

    if (*req_type == -1)
	if (*req_from == -1)
	    matcher = TCMP_MATCH_ANY;
	else
	    matcher = TCMP_MATCH_SENDER;
    else if (*req_from == -1)
	matcher = TCMP_MATCH_TYPE;
    else
	matcher = TCMP_MATCH_BOTH;
    tcmpstat = tcmp_receive(matcher, req_from, req_type,
			    TCMP_NOBLOCK | TCMP_NOCOPY | TCMP_NODEQUEUE,
			    &len_rcvd, &msg);
    if (tcmpstat == TCMP_SUCCESS)
	return(P4_TRUE);
    else
	return(P4_FALSE);
}

int MD_tcmp_send(type, from, to, msg, len, data_type, ack_req)
int type, from, to, len, data_type, ack_req;
char *msg;
{
    int sendflags;
    tcmp_status tcmpstat;

    if (ack_req & P4_ACK_REQ_MASK)
	sendflags = 0;
    else
	sendflags = TCMP_NOBLOCK;
    tcmpstat = tcmp_send(to, type, sendflags, len, msg);
    if (tcmpstat != TCMP_SUCCESS)
	p4_error("bad status on tcmp_send = ",tcmpstat);
    return(0);
}

struct p4_msg *MD_tcmp_recv()
{
    int type, from, len;
    struct p4_msg *msg;
    tcmp_status tcmpstat;

    tcmpstat = tcmp_receive(TCMP_MATCH_ANY,&from,&type,TCMP_NOCOPY,&len,&msg);
    if (tcmpstat != TCMP_SUCCESS)
	p4_error("bad tcmp status on receive = ", tcmpstat);

    return (msg);
}

#endif

int data_representation( char *machine_type )
{
    if (strcmp(machine_type, "SUN") == 0)             return 1;
    if (strcmp(machine_type, "HP") == 0)              return 1;
    if (strcmp(machine_type, "RS6000") == 0)          return 1;
    if (strcmp(machine_type, "SGI") == 0)             return 1;
    if (strcmp(machine_type, "NEXT") == 0)            return 1;
    if (strcmp(machine_type, "CM5") == 0)             return 1;
    if (strcmp(machine_type, "SYMMETRY") == 0)        return 2;
    if (strcmp(machine_type, "SYMMETRY_PTX") == 0)    return 2;
    if (strcmp(machine_type, "SUN386I") == 0)         return 2;
#ifdef WORDS_BIGENDIAN
    /* These are ported to some non-IA32 system */
    if (strcmp(machine_type, "LINUX") == 0)           return 21;
    if (strcmp(machine_type, "FREEBSD") == 0)         return 22;
    if (strcmp(machine_type, "NETBSD") == 0)          return 23;
#else
    if (strcmp(machine_type, "LINUX") == 0)           return 2;
    if (strcmp(machine_type, "FREEBSD") == 0)         return 2;
    if (strcmp(machine_type, "NETBSD") == 0)          return 2;
#endif
    if (strcmp(machine_type, "I86_SOLARIS") == 0)     return 2;
    if (strcmp(machine_type, "DEC5000") == 0)         return 3;
    if (strcmp(machine_type, "IBM3090") == 0)         return 4;
    if (strcmp(machine_type, "TITAN") == 0)           return 5;
    if (strcmp(machine_type, "FX8") == 0)             return 6;
    if (strcmp(machine_type, "FX2800") == 0)          return 7;
    if (strcmp(machine_type, "FX2800_SWITCH") == 0)   return 7;
    if (strcmp(machine_type, "IPSC860") == 0)         return 8;
    if (strcmp(machine_type, "IPSC860_SOCKETS") == 0) return 8;
    if (strcmp(machine_type, "DELTA") == 0)           return 8;
    if (strcmp(machine_type, "BALANCE") == 0)         return 12;
    if (strcmp(machine_type, "MULTIMAX") == 0)        return 15;
    if (strcmp(machine_type, "CRAY") == 0)            return 16;
    if (strcmp(machine_type, "GP_1000") == 0)         return 17;
    if (strcmp(machine_type, "TC_2000") == 0)         return 18;
    if (strcmp(machine_type, "TC_2000_TCMP") == 0)    return 18;
    if (strcmp(machine_type, "KSR") == 0)             return 19;
    if (strcmp(machine_type, "NCUBE") == 0)           return 20;

    if (strcmp(machine_type, "LINUX_PPC") == 0)       return 24;
    if (strcmp(machine_type, "LINUX_ALPHA") == 0)     return 25;
    p4_dprintf("invalid machine type=:%s:\n",machine_type);
    p4_error("data_representation: invalid machine type",0);
    return(-1);
}

#if defined(FREEBSD) && !defined(HAVE_XDR_FLOAT)
/* As of release 1.1, FreeBSD leaves out the xdr_float.c files from
   /usr/lib/libc.a.  This is the source code from FreeBSD 1.1.  This
   has been tested, and seems to work well on all but NaN's */
#include "xdr_float.c"
#endif