p4_rm.c

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

#include "p4.h"
#include "p4_sys.h"

/*
  Defining REDIRECT_OUTPUT allows you to redirect output from
  processes as follows.
  Defining OUT_TO_TERM tty allows you to redirect the
  output from various processes to separate windows.
  */

#if defined(REDIRECT_OUTPUT)
#if defined(OUT_TO_TERM)
#define P4_OUTFILE "/dev/ttyp4"
#else
#define P4_OUTFILE "/tmp/p4out"
#endif
#endif

static int rm_num;
static int rm_flag;

int rm_start( int *argc, char **argv)
{
    int bm_fd, bm_port;
    char *s,*bm_host;
    extern char whoami_p4[];
    struct net_initial_handshake hs;
/*     struct slave_listener_msg lmsg; */
    struct bm_rm_msg msg;
    int type, rc, numslaves;
#if defined(SP1_EUIH)
    int len;
#endif
#if defined(NCUBE)
    int unused_flag;
#endif
#if defined(IPSC860)  ||  defined(CM5)  ||  defined(NCUBE)  ||  defined(SP1_EUI) || defined(SP1_EUIH)
    int i;
#endif
    int conn_retries;
    struct p4_global_data *g;
    char outfile[P4_MAX_PGM_LEN];

    trap_sig_errs(); 		/* Errors can happen any time */

    sprintf(whoami_p4, "rm_%d", (int)getpid());
    p4_dprintfl(20,"remote master starting, my p4 id is %s\n",whoami_p4);

#if defined(REDIRECT_OUTPUT)
    freopen(P4_OUTFILE, "w", stdout);
    freopen(P4_OUTFILE, "w", stderr);
#endif

    conn_retries = 5;
    if (execer_mynodenum)
    {
        bm_host = execer_masthost;
        bm_port = execer_mastport;
	conn_retries = 100;
    }
    else
    {
	if (*argc < 4)
	    p4_error("Invalid arguments to remote_master", *argc);

	bm_host = argv[1];
	bm_port = atoi(argv[2]);
    }

    bm_fd = net_conn_to_listener(bm_host, bm_port, conn_retries);
    if (bm_fd < 0)
	p4_error("rm_start: net_conn_to_listener failed", bm_port);

    net_recv(bm_fd, &hs, sizeof(hs));
    hs.pid = (int) htonl(getpid());
    hs.rm_num = (int) htonl(execer_mynodenum);  /* only used with dqs */
    net_send(bm_fd, &hs, sizeof(hs), P4_FALSE);

#   ifdef SYSV_IPC
    sysv_num_shmids = 0;
    sysv_shmid[0]  = -1;
    sysv_semid0    = -1;
    sysv_semid0 = init_sysv_semset(0);
#   endif

    /* Get the initialization information from the bm */

    rc = net_recv(bm_fd, &msg, sizeof(msg));
    if (rc == PRECV_EOF)
	p4_error("rm_start: got EOF on net_recv", bm_fd);
    type = p4_n_to_i(msg.type);
    if (type != INITIAL_INFO)
	p4_error("rm_start: unknown type, expecting INITIAL_INFO, type=", type);
    if (strcmp(msg.version,P4_PATCHLEVEL) != 0)
    {
	p4_dprintf("my version is %s, received %s as version\n",
		   P4_PATCHLEVEL,msg.version);
	p4_error("version does not match master \n",0);
    }

    /* choose working directory */
    if (strlen(msg.wdir) && !chdir(msg.wdir))
    {
	p4_dprintfl(90, "working directory set to %s\n", msg.wdir);
    }
    else
    {
	if ((s = (char *) rindex(msg.pgm,'/'))  !=  NULL)
	{
	    *s = '\0';  /* chg to directory name only */
	    chdir(msg.pgm);
	}
    }
    globmemsize = p4_n_to_i(msg.memsize);
    logging_flag = p4_n_to_i(msg.logging_flag);
    if (logging_flag)
	ALOG_ENABLE;
    else
	ALOG_DISABLE;

    MD_initmem(globmemsize);
    alloc_global();  /* sets p4_global */
    g = p4_global;
    p4_local = alloc_local_rm();
    g->local_communication_only = P4_FALSE;
    g->num_in_proctable = p4_n_to_i(msg.numinproctab);
    numslaves = p4_n_to_i(msg.numslaves);
    rm_num = p4_n_to_i(msg.rm_num);
    p4_debug_level = p4_n_to_i(msg.debug_level);
    strncpy(outfile, msg.outfile, P4_MAX_PGM_LEN);
    outfile[P4_MAX_PGM_LEN-1] = 0;
    strcpy(p4_global->application_id, msg.application_id);
    p4_dprintfl(90, "got numslaves=%d outfile=%s rm_num=%d dbglvl=%d appid=%s\n",
		numslaves, outfile, rm_num, p4_debug_level, msg.application_id);

    MD_initenv();
    usc_init();
    init_usclock();

    if (*outfile)
    {
	freopen(outfile, "w", stdout);
	freopen(outfile, "w", stderr);
    }

#ifndef THREAD_LISTENER
    SIGNAL_P4(LISTENER_ATTN_SIGNAL, handle_connection_interrupt);
#endif
    p4_lock(&g->slave_lock);
    create_rm_processes(numslaves, bm_fd);
    if (!rm_flag) /* I am not rm; was forked in create_rm_processes */
	return(0);

    /* Grab the whole proc table from the bm */
    p4_dprintfl(90, "receiving proc table\n");
    receive_proc_table(bm_fd);

    /* let local slaves use proctable to identify themselves */
    p4_unlock(&g->slave_lock);

    sprintf(whoami_p4, "rm_%d_%d", rm_num, (int)getpid());
    p4_local->my_id = p4_get_my_id_from_proc();

    p4_global->low_cluster_id = 
	p4_local->my_id - p4_global->proctable[p4_local->my_id].slave_idx;
    p4_global->hi_cluster_id = 
	p4_global->low_cluster_id + p4_global->local_slave_count;

    setup_conntab();

    if (p4_local->conntab[0].type == CONN_REMOTE_SWITCH)
    {
	p4_local->conntab[0].switch_port = p4_global->proctable[0].switch_port;
	p4_local->conntab[0].port = bm_fd;
    }
    else if (p4_local->conntab[0].type == CONN_REMOTE_NON_EST)
    {
	p4_local->conntab[0].type = CONN_REMOTE_EST;
	p4_local->conntab[0].port = bm_fd;
	p4_local->conntab[0].same_data_rep =
	    same_data_representation(p4_local->my_id,0);
    }
    else
    {
	p4_error("rm_start: invalid conn type in conntab ",
		 p4_local->conntab[0].type);
    }

    sprintf(whoami_p4, "p%d_%d", p4_get_my_id(), (int)getpid());


#if defined(IPSC860)  ||  defined(CM5)  ||  defined(NCUBE)  ||  defined(SP1_EUI) || defined(SP1_EUIH)
    for (i = 1; i < numslaves; i++)
    {
#       if defined(IPSC860)
	csend((long) INITIAL_INFO, &msg, (long) sizeof(struct bm_rm_msg), 
              (long) i, (long) NODE_PID);
	csend((long) INITIAL_INFO, p4_global->proctable, 
              (long) sizeof(p4_global->proctable), (long) i, (long) NODE_PID);
#       endif
#       if defined(CM5)
	CMMD_send_noblock(i, INITIAL_INFO,  &msg,sizeof(struct bm_rm_msg));
	CMMD_send_noblock(i, INITIAL_INFO,  p4_global->proctable, 
                          sizeof(p4_global->proctable));
#       endif
#       if defined(NCUBE)
	nwrite(&msg, sizeof(struct bm_rm_msg), i, INITIAL_INFO, &unused_flag);
	nwrite(p4_global->proctable, sizeof(p4_global->proctable), i, INITIAL_INFO, &unused_flag);
#       endif
#       if defined(SP1_EUI)
	mpc_bsend(&msg, sizeof(struct bm_rm_msg), i, INITIAL_INFO);
	mpc_bsend(p4_global->proctable, sizeof(p4_global->proctable), i, INITIAL_INFO);
#       endif
#       if defined(SP1_EUIH)
	len  = sizeof(struct bm_rm_msg);
	type = INITIAL_INFO;
	mp_bsend(&msg, &len, &i, &type);
	len  = sizeof(p4_global->proctable);
	type = INITIAL_INFO;
	mp_bsend(p4_global->proctable, &len, &i, &type);
#       endif
    }
#endif

    /* 
       sync with local slaves thus insuring that they have the proctable before 
       syncing with bm (this keeps bm and its slaves  from interrupting the local 
       processes too early; then re-sync with local slaves (thus permitting them to 
       interrupt remotes)
    */
    p4_barrier(&(p4_global->cluster_barrier),p4_num_cluster_ids());
    
    msg.type = p4_i_to_n(SYNC_MSG);
    net_send(bm_fd, &msg, sizeof(msg), P4_FALSE);
    msg.type = -1;  /* reset to verify type received next */
    rc = net_recv(bm_fd, &msg, sizeof(msg));
    type = p4_n_to_i(msg.type);
    if (type != SYNC_MSG)
	p4_error("rm_start: unknown type, expecting SYNC_MSG, type=", type);
    
    p4_barrier(&(p4_global->cluster_barrier),p4_num_cluster_ids());

    return(0);
}


P4VOID create_rm_processes(int nslaves, int bm_fd)
{
    struct p4_global_data *g = p4_global;
    int end_1, end_2, slave_pid, listener_pid;
    int slave_idx, listener_port, listener_fd;
    char rm_host[100];
    int rm_switch_port;
    struct bm_rm_msg bm_msg;
#   if defined(IPSC860)  ||  defined(CM5)  ||  defined(NCUBE)  ||  defined(SP1_EUI) || defined(SP1_EUIH)
    int i, from, type, len, unused_flag;
#endif
#   if defined(THREAD_LISTENER)
    p4_thread_t trc;
#   endif

#   if !defined(IPSC860)  &&  !defined(CM5)  &&  !defined(NCUBE)  &&  !defined(SP1_EUI) && !defined(SP1_EUIH)
    if (nslaves > P4_MAX_MSG_QUEUES)
	p4_error("create_rm_processes: more slaves than msg queues \n", nslaves);
#   endif

    rm_flag = 0;  /* set below for the remote master */
    /*
     * Allocate the listener's local data area, since this process will
     * eventually become the listener.
     */

    /* nslaves is total number of processes on the remote machine; this
     * is not the case in the big master code, though. */
    listener_info = alloc_listener_info(nslaves);

    net_setup_anon_listener(MAX_P4_CONN_BACKLOG, &listener_port, &listener_fd);
    listener_info->listening_fd = listener_fd;

    p4_dprintfl(70, "created listener on port %d fd %d\n", listener_port,
		listener_fd);

    /* Send off the listener info to the bm */
    bm_msg.type = p4_i_to_n(REMOTE_LISTENER_INFO);