tcp_module_init.c

fortran并行计算包

                              IPPROTO_TCP,
                              TCP_NODELAY,
                              &option,
                              sizeof(int));
            MPIU_ERR_CHKANDJUMP1 (ret == -1, mpi_errno, MPI_ERR_OTHER, "**setsockopt", "**setsockopt %s", strerror (errno));
 	    getsockopt(nodes[grank].desc,IPPROTO_TCP,TCP_NODELAY,&option2,&size);

	    option = 2 * MPID_NEM_CELL_PAYLOAD_LEN ;
	    ret = setsockopt( nodes[grank].desc,
                              SOL_SOCKET,
                              SO_RCVBUF,
                              &option,
                              sizeof(int));
            MPIU_ERR_CHKANDJUMP1 (ret == -1, mpi_errno, MPI_ERR_OTHER, "**setsockopt", "**setsockopt %s", strerror (errno));
	    getsockopt(nodes[grank].desc,SOL_SOCKET,SO_RCVBUF,&option2,&size);

	    ret = setsockopt( nodes[grank].desc,
                              SOL_SOCKET,
                              SO_SNDBUF,
                              &option,
                              sizeof(int));
            MPIU_ERR_CHKANDJUMP1 (ret == -1, mpi_errno, MPI_ERR_OTHER, "**setsockopt", "**setsockopt %s", strerror (errno));
	    getsockopt(nodes[grank].desc,SOL_SOCKET,SO_SNDBUF,&option2,&size);
	
	    /* TCP_MAXSEG is may not be defined if we are enforcing
	       strict POSIX_C_SOURCE (on OSX, for example) */
#ifdef TCP_MAXSEG
	    setsockopt( nodes[grank].desc, IPPROTO_TCP,TCP_MAXSEG,&option,sizeof(int));
	    getsockopt(nodes[grank].desc,IPPROTO_TCP,TCP_MAXSEG,&option2,&size);
#endif
	}
    }
    (MPID_nem_tcp_internal_vars.max_fd)++;

    MPIU_CHKPMEM_COMMIT();
 fn_exit:
    return mpi_errno;
 fn_fail:
    MPIU_CHKPMEM_REAP();
    goto fn_exit;
}


/*
   int
   MPID_nem_tcp_module_init(MPID_nem_queue_ptr_t proc_recv_queue, MPID_nem_queue_ptr_t proc_free_queue, MPID_nem_cell_ptr_t proc_elements, int num_proc_elements,
                            MPID_nem_cell_ptr_t module_elements, int num_module_elements,
                            MPID_nem_queue_ptr_t *module_free_queue)

   IN
       proc_recv_queue -- main recv queue for the process
       proc_free_queue -- main free queueu for the process
       proc_elements -- pointer to the process' queue elements
       num_proc_elements -- number of process' queue elements
       module_elements -- pointer to queue elements to be used by this module
       num_module_elements -- number of queue elements for this module
       ckpt_restart -- true if this is a restart from a checkpoint.  In a restart, the network needs to be brought up again, but
                       we want to keep things like sequence numbers.
   OUT
       free_queue -- pointer to the free queue for this module.  The process will return elements to
                     this queue
*/

#undef FUNCNAME
#define FUNCNAME MPID_nem_tcp_module_init
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int
MPID_nem_tcp_module_init (MPID_nem_queue_ptr_t  proc_recv_queue,
			  MPID_nem_queue_ptr_t  proc_free_queue,
			  MPID_nem_cell_ptr_t    proc_elements,
			  int num_proc_elements,
			  MPID_nem_cell_ptr_t    module_elements,
			  int num_module_elements,
			  MPID_nem_queue_ptr_t *module_free_queue,
			  int ckpt_restart, MPIDI_PG_t *pg_p, int pg_rank,
			  char **bc_val_p, int *val_max_sz_p)
{
    int mpi_errno = MPI_SUCCESS;
    int index;
    MPIU_CHKPMEM_DECL(1);
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_MODULE_INIT);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_MODULE_INIT);

    MPID_nem_tcp_internal_vars.n_pending_send  = 0;
    MPID_nem_tcp_internal_vars.n_pending_recv  = 0;
    MPID_nem_tcp_internal_vars.outstanding     = 0;
    MPID_nem_tcp_internal_vars.poll_freq       = TCP_POLL_FREQ_NO;
    MPID_nem_tcp_internal_vars.old_poll_freq   = TCP_POLL_FREQ_NO;

    MPIU_CHKPMEM_MALLOC (MPID_nem_tcp_internal_vars.n_pending_sends, int *, MPID_nem_mem_region.num_procs * sizeof (int), mpi_errno, "pending sends");
    for(index = 0 ; index < MPID_nem_mem_region.num_procs ; index++)
    {
	MPID_nem_tcp_internal_vars.n_pending_sends[index] = 0;
    }

    if( MPID_nem_mem_region.ext_procs > 0)
    {
	mpi_errno = init_tcp (pg_p);
        if (mpi_errno) MPIU_ERR_POP (mpi_errno);

	if(MPID_nem_mem_region.num_local == 0)
	    MPID_nem_tcp_internal_vars.poll_freq = TCP_POLL_FREQ_ALONE ;
	else
	    MPID_nem_tcp_internal_vars.poll_freq = TCP_POLL_FREQ_MULTI ;
	MPID_nem_tcp_internal_vars.old_poll_freq = MPID_nem_tcp_internal_vars.poll_freq;	
    }

    MPID_nem_process_recv_queue = proc_recv_queue;
    MPID_nem_process_free_queue = proc_free_queue;

    MPID_nem_module_tcp_free_queue = &_free_queue;

    MPID_nem_queue_init (MPID_nem_module_tcp_free_queue);

    for (index = 0; index < num_module_elements; ++index)
    {
	MPID_nem_queue_enqueue (MPID_nem_module_tcp_free_queue, &module_elements[index]);
    }

    *module_free_queue = MPID_nem_module_tcp_free_queue;

    MPIU_CHKPMEM_COMMIT();
 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_MODULE_INIT);
    return mpi_errno;
 fn_fail:
    MPIU_CHKPMEM_REAP();
    goto fn_exit;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_tcp_module_get_business_card
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int
MPID_nem_tcp_module_get_business_card (int my_rank, char **bc_val_p, int *val_max_sz_p)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_MODULE_GET_BUSINESS_CARD);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_MODULE_GET_BUSINESS_CARD);

    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**notimpl", 0);

    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_MODULE_GET_BUSINESS_CARD);
    return mpi_errno;
}


#undef FUNCNAME
#define FUNCNAME MPID_nem_tcp_module_connect_to_root
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int
MPID_nem_tcp_module_connect_to_root (const char *business_card, MPIDI_VC_t *new_vc)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_MODULE_CONNECT_TO_ROOT);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_MODULE_CONNECT_TO_ROOT);

    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**notimpl", 0);

    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_MODULE_CONNECT_TO_ROOT);
    return mpi_errno;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_tcp_module_vc_init
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int
MPID_nem_tcp_module_vc_init (MPIDI_VC_t *vc, const char *business_card)
{
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_MODULE_VC_INIT);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_MODULE_VC_INIT);

    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_MODULE_VC_INIT);
    return MPI_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_tcp_module_vc_destroy
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPID_nem_tcp_module_vc_destroy(MPIDI_VC_t *vc)
{
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_MODULE_VC_DESTROY);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_MODULE_VC_DESTROY);

    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_MODULE_VC_DESTROY);
    return MPI_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME MPID_nem_tcp_module_vc_terminate
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPID_nem_tcp_module_vc_terminate (MPIDI_VC_t *vc)
{
    MPIDI_STATE_DECL(MPID_STATE_MPID_NEM_TCP_MODULE_VC_TERMINATE);

    MPIDI_FUNC_ENTER(MPID_STATE_MPID_NEM_TCP_MODULE_VC_TERMINATE);

    MPIDI_FUNC_EXIT(MPID_STATE_MPID_NEM_TCP_MODULE_VC_TERMINATE);
    return MPI_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME getSockInterfaceAddr
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int getSockInterfaceAddr( int myRank, char *ifname, int maxIfname)
{
    char *ifname_string;
    int mpi_errno = MPI_SUCCESS;

    /* Check for the name supplied through an environment variable */
    ifname_string = getenv("MPICH_INTERFACE_HOSTNAME");
    if (!ifname_string) {
	/* See if there is a per-process name for the interfaces (e.g.,
	   the process manager only delievers the same values for the
	   environment to each process */
	char namebuf[1024];
	MPIU_Snprintf( namebuf, sizeof(namebuf),
		       "MPICH_INTERFACE_HOSTNAME_R%d", myRank );
	ifname_string = getenv( namebuf );
	if (dbg_ifname && ifname_string) {
	    fprintf( stdout, "Found interface name %s from %s\n",
		    ifname_string, namebuf );
	    fflush( stdout );
	}
    }
    else if (dbg_ifname) {
	fprintf( stdout,
		 "Found interface name %s from MPICH_INTERFACE_HOSTNAME\n",
		 ifname_string );
	fflush( stdout );
    }

    if (!ifname_string) {
	int len;

	/* If we have nothing, then use the host name */
	mpi_errno = MPID_Get_processor_name(ifname, maxIfname, &len );
        if (mpi_errno) MPIU_ERR_POP(mpi_errno);

	ifname_string = ifname;

	/* If we didn't find a specific name, then try to get an IP address
	   directly from the available interfaces, if that is supported on
	   this platform.  Otherwise, we'll drop into the next step that uses
	   the ifname */
    }
    else {
	/* Copy this name into the output name */
	MPIU_Strncpy( ifname, ifname_string, maxIfname );
    }

 fn_exit:
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}