p4_bm.c

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

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

int bm_start(int *argc, char **argv)
{
    int bm_switch_port;
    char *s,pgmname[P4_MAX_PGM_LEN];

    sprintf(whoami_p4, "p0_%d", (int)getpid());
    p4_dprintfl(90,"entering bm_start\n");

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

    /* choose a working directory */
    if (strlen(p4_wd) && !chdir(p4_wd))
    {
	p4_dprintfl(90, "working directory set to %s\n", p4_wd);
    }
    else
    {
	strncpy(pgmname,argv[0],P4_MAX_PGM_LEN);
	pgmname[P4_MAX_PGM_LEN-1] = 0;
	if ((s = (char *) rindex(pgmname,'/'))  !=  NULL)
	{
	    *s = '\0';  /* chg to directory name only */
	    chdir(pgmname);
	}
    }

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

    MD_initmem(globmemsize);
    alloc_global();  /* sets p4_global */

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

    p4_local = alloc_local_bm();
    if (p4_local == NULL)
	p4_error("p4_initenv: alloc_local_bm failed\n", 0);

    MD_initenv();
#ifdef CAN_DO_SWITCH_MSGS
    bm_switch_port = getswport(p4_global->my_host_name);
#else
    bm_switch_port = -1;
#endif
    usc_init();
    init_usclock();

    /* big master installing himself */
    install_in_proctable(0, (-1), getpid(), p4_global->my_host_name, 
			 p4_global->my_host_name, 
			 0, P4_MACHINE_TYPE, bm_switch_port);

    p4_local->my_id = 0;

    if (logging_flag)
	ALOG_ENABLE;
    else
	ALOG_DISABLE;

    return (0);
}

int p4_create_procgroup()
{

    p4_dprintfl(90,"entering p4_create_procgroup\n");
    if (p4_local->my_id != 0) {
	p4_local->procgroup = 0;
	return(0);
    }
    if (execer_pg)
    {
	p4_local->procgroup = execer_pg;
    }
    else
    {
	if ((p4_local->procgroup = read_procgroup()) == NULL)
	    return (-1);
    }
    p4_startup(p4_local->procgroup);
    return(0);
}

/*
 * Start both the listener and the local and remote processes.
 */
int p4_startup( struct p4_procgroup *pg )
{
    int nslaves;
    int listener_port, listener_fd;

    p4_dprintfl(90,"entering p4_startup\n");

    if (p4_global == NULL)
	p4_error("p4 not initialized; perhaps p4_initenv not called",0);

/* On some systems (SGI IRIX 6), process exit sometimes kills all processes
   in the process GROUP.  This code attempts to fix that.  
   We DON'T do it if stdin (0) is connected to a terminal, because that
   disconnects the process from the terminal.
 */
#if defined(HAVE_SETSID) && defined(HAVE_ISATTY) && defined(SET_NEW_PGRP)
if (!isatty(0)) {
    pid_t rc;
    rc = setsid();
    if (rc < 0) {
	p4_dprintfl( 90, "Could not create new process group\n" );
	}
    else {
	p4_dprintfl( 80, "Created new process group %d\n", rc );
	}
    }
else {
	p4_dprintfl( 80, 
         "Did not create new process group because isatty returned true\n" );
    }
#endif

    procgroup_to_proctable(pg);
    if (pg->num_entries > 1)
	p4_global->local_communication_only = P4_FALSE;

#   ifdef CAN_DO_SOCKET_MSGS
    if (!p4_global->local_communication_only)
    {
	net_setup_anon_listener(MAX_P4_CONN_BACKLOG, &listener_port, 
				&listener_fd); 
	p4_global->listener_port = listener_port;
	p4_global->listener_fd = listener_fd;
	p4_dprintfl(90, "setup listener on port %d fd %d\n",
		    listener_port, listener_fd);
	p4_global->proctable[0].port = listener_port;
#ifndef THREAD_LISTENER
	SIGNAL_P4(LISTENER_ATTN_SIGNAL, handle_connection_interrupt);
#endif
    }
    else
	p4_global->listener_fd = -1;
#   endif

    setup_conntab();

    p4_lock(&p4_global->slave_lock);
    if ((nslaves = create_bm_processes(pg)) < 0)
	return (-1);
    if (!p4_am_i_cluster_master())  /* I was forked in create_bm_processes */
	return(0);

#   ifdef CAN_DO_SOCKET_MSGS
    if (create_remote_processes(pg) < 0)
	return (-1);
#   endif

    /* let local slaves use proc table to identify themselves */
    p4_unlock(&p4_global->slave_lock); 

    send_proc_table();  /* to remote masters */

#   if defined(IPSC860)  ||  defined(CM5)     ||  defined(NCUBE)  \
                         ||  defined(SP1_EUI) ||  defined(SP1_EUIH)
     {
    struct bm_rm_msg bm_msg;
    struct p4_procgroup_entry *local_pg;
    int len, to, type;
    int i, unused_flag;

    /* send initial info and proctable to local slaves */
    /* must use p4_i_to_n procs because node slave 
       does not know if the msg is forwarded from bm */
    local_pg = &(pg->entries[0]);
    bm_msg.type = p4_i_to_n(INITIAL_INFO);
    bm_msg.numinproctab = p4_i_to_n(p4_global->num_in_proctable);
    bm_msg.numslaves = p4_i_to_n(local_pg->numslaves_in_group);
    bm_msg.debug_level = p4_i_to_n(p4_remote_debug_level);
    bm_msg.memsize = p4_i_to_n(globmemsize);
    bm_msg.logging_flag = p4_i_to_n(logging_flag);
    strcpy(bm_msg.application_id, p4_global->application_id);
    strcpy(bm_msg.version, P4_PATCHLEVEL);
    if (strlen( local_pg->slave_full_pathname ) >= P4_MAX_PGM_LEN) {
	p4_error("Program name is too long, must be less than", 
		 P4_MAX_PGM_LEN);
    }
    strcpy(bm_msg.pgm, local_pg->slave_full_pathname);
    strcpy(bm_msg.wdir, p4_wd);
    for (i = 1; i <= nslaves; i++)
    {
	p4_dprintfl(90,"sending initinfo to slave %d of %d\n",i,nslaves);
#       if defined(IPSC860)
	csend((long) INITIAL_INFO, &bm_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, &bm_msg, sizeof(struct bm_rm_msg));
	CMMD_send_noblock(i, INITIAL_INFO, p4_global->proctable, sizeof(p4_global->proctable));
#       endif
#       if defined(NCUBE)
	nwrite(&bm_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(&bm_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);
	to   = i;
	type = INITIAL_INFO; 
	mp_bsend(&bm_msg, &len, &to, &type); 
	len  = sizeof(p4_global->proctable);
	mp_bsend(p4_global->proctable, &len, &to, &type);
#       endif
	p4_dprintfl(90,"sent initinfo to slave %d of %d\n",i,nslaves);
    }
     }   /* End of local declarations */
#   endif

    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 + 1;

    /* 
       sync with local slaves thus insuring that they have the proctable before 
       syncing with remotes (this keeps remotes 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());
    /* 
       NEED A SYNC WITH LOCALS THAT DOES A BARRIER WITH PROCS THAT SHARE
       MEMORY AND MP BARRIER WITH OTHER "LOCAL" PROCESSES 
    */
    sync_with_remotes();
    p4_barrier(&(p4_global->cluster_barrier),p4_num_cluster_ids());


    return (0);
}

int create_bm_processes( struct p4_procgroup *pg )
{
    struct p4_procgroup_entry *local_pg;
    int nslaves, end_1, end_2;
    int slave_pid, listener_pid = -1;
    int slave_idx, listener_fd = -1;
#   if defined(IPSC860)  ||  defined(CM5)  ||  defined(NCUBE)  ||  defined(SP1_EUI) || defined(SP1_EUIH)
    /* Message passing systems require additional information */
    struct bm_rm_msg bm_msg;
    int i;
    int port, switch_port, type, len, from, unused_flag
#endif
#   if defined(THREAD_LISTENER)
    p4_thread_t trc;
#   endif

    p4_dprintfl(90,"entering create_bm_processes\n");
    local_pg = &(pg->entries[0]);

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

/* alloc listener local data since this proc will eventually become listener */
#   if defined(CAN_DO_SOCKET_MSGS)  &&  !defined(NO_LISTENER)
    if (!(p4_global->local_communication_only))
    {
	listener_fd = p4_global->listener_fd;
	listener_info = alloc_listener_info(nslaves+1);
	listener_info->listening_fd = listener_fd;
    }
#   endif

#   ifdef TCMP
    tcmp_init(NULL,p4_get_my_cluster_id(),shmem_getclunid());
#   endif

#   if defined(IPSC860)  ||  defined(CM5)  ||  defined(NCUBE)  ||  defined(SP1_EUI) || defined(SP1_EUIH)
    for (i = 1; i <= nslaves; i++)
    {
	p4_dprintfl(90,"doing initial sync with local slave %d\n",i);
#       if defined(IPSC860)
	csend((long) SYNC_MSG, &bm_msg, (long) sizeof(struct bm_rm_msg), 
	      (long) i, (long) NODE_PID);
	crecv(INITIAL_INFO, &bm_msg, (long) sizeof(struct bm_rm_msg));
#       endif /* IPSC860 */
#       if defined(CM5)
	CMMD_send_noblock(i, SYNC_MSG, &bm_msg, sizeof(struct bm_rm_msg));
        CMMD_receive(CMMD_ANY_NODE, INITIAL_INFO, (void *) &bm_msg, sizeof(struct bm_rm_msg));
#       endif /* CM5 */
#       if defined(NCUBE)
	nwrite(&bm_msg, sizeof(struct bm_rm_msg), i, SYNC_MSG, &unused_flag);
        from = NCUBE_ANY_NODE;
        type = INITIAL_INFO;
        nread(&bm_msg, sizeof(struct bm_rm_msg), &from, &type, &unused_flag);
#       endif /* NCUBE */
#       if defined(SP1_EUI)
	mpc_bsend(&bm_msg, sizeof(struct bm_rm_msg), i, SYNC_MSG);
        from = ANY_P4TYPE_EUI;
        type = INITIAL_INFO;
        mpc_brecv(&bm_msg, sizeof(struct bm_rm_msg), &from, &type, &unused_flag);
#       endif /*SP1_EUI */
#       if defined(SP1_EUIH)
	len = sizeof(struct bm_rm_msg);
	type = SYNC_MSG;
	mp_bsend(&bm_msg, &len, &i, &type);
        from = ANY_P4TYPE_EUIH;
        type = INITIAL_INFO;
	len  = sizeof(struct bm_rm_msg);
        mp_brecv(&bm_msg, &len, &from, &type, &unused_flag);
#       endif /* SP1_EUIH */
	port = p4_n_to_i(bm_msg.port);
	slave_idx = p4_n_to_i(bm_msg.slave_idx);
	slave_pid = p4_n_to_i(bm_msg.slave_pid);
	switch_port = p4_n_to_i(bm_msg.switch_port);
	/* big master installing local slaves */
	install_in_proctable(0, port, slave_pid, bm_msg.host_name, 
			     bm_msg.local_name, 
			     slave_idx, P4_MACHINE_TYPE, switch_port);
	p4_global->local_slave_count++;
    }
#   else
#if defined(SUN_SOLARIS)
/*****  Shyam code, removed by RL
    { processorid_t proc = 0;
      if(p_online(proc,P_STATUS) != P_ONLINE)
	printf("Could not bind parent to processor 0\n");
      else
	{
	  processor_bind(P_PID,P_MYID,proc, &proc);
	  printf("Bound parent to processor 0 , previous binding was %d\n",
		 proc);
	}
    }
*****/
#endif
    /* The following creates the slaves for systems that use fork to 
       create the *additional* processes (we're the initial process) */
    for (slave_idx = 1; slave_idx <= nslaves; slave_idx++)
    {
	p4_dprintfl(20, "creating local slave %d of %d\n",slave_idx,nslaves);
#       if defined(CAN_DO_SOCKET_MSGS)  &&  !defined(NO_LISTENER)
        if (!(p4_global->local_communication_only)) {
	    get_pipe(&end_1, &end_2);
	    listener_info->slave_fd[slave_idx] = end_2;
	    slave_pid = fork_p4();
	    listener_info->slave_pid[slave_idx] = slave_pid;
	}
	else {
	    slave_pid = fork_p4();
	}
#       else
	slave_pid = fork_p4();
#       endif
	if (slave_pid < 0)
	    p4_error("create_bm_processes fork", slave_pid);
	else {
	    if (slave_pid)