btl_tcp_component.c

来自「MPI stands for the Message Passing Inter」· C语言 代码 · 共 642 行 · 第 1/2 页

 * (1) all interfaces specified by the user
 * (2) all available interfaces 
 * (3) all available interfaces except for those excluded by the user
 */

static int mca_btl_tcp_component_create_instances(void)
{
    int if_count = opal_ifcount();
    int if_index;
    char **include;
    char **exclude;
    char **argv;

    if(if_count <= 0)
        return OMPI_ERROR;

    /* allocate memory for btls */
    mca_btl_tcp_component.tcp_btls = (mca_btl_tcp_module_t **)malloc(if_count * sizeof(mca_btl_tcp_module_t*));
    if(NULL == mca_btl_tcp_component.tcp_btls)
        return OMPI_ERR_OUT_OF_RESOURCE;

    /* if the user specified an interface list - use these exclusively */
    argv = include = opal_argv_split(mca_btl_tcp_component.tcp_if_include,',');
    while(argv && *argv) {
        char* if_name = *argv;
        int if_index = opal_ifnametoindex(if_name);
        if(if_index < 0) {
            BTL_ERROR(("invalid interface \"%s\"", if_name));
        } else {
            mca_btl_tcp_create(if_index, if_name);
        }
        argv++;
    }
    opal_argv_free(include);
    if(mca_btl_tcp_component.tcp_num_btls)
        return OMPI_SUCCESS;

    /* if the interface list was not specified by the user, create 
     * a BTL for each interface that was not excluded.
    */
    exclude = opal_argv_split(mca_btl_tcp_component.tcp_if_exclude,',');
    for(if_index = opal_ifbegin(); if_index >= 0; if_index = opal_ifnext(if_index)) {
        char if_name[32];
        opal_ifindextoname(if_index, if_name, sizeof(if_name));

        /* check to see if this interface exists in the exclude list */
        if(opal_ifcount() > 1) {
            argv = exclude;
            while(argv && *argv) {
                if(strncmp(*argv,if_name,strlen(*argv)) == 0)
                    break;
                argv++;
            }
            /* if this interface was not found in the excluded list - create a BTL */
            if(argv == 0 || *argv == 0) {
                mca_btl_tcp_create(if_index, if_name);
            }
        } else {
            mca_btl_tcp_create(if_index, if_name);
        }
    }
    opal_argv_free(exclude);
    return OMPI_SUCCESS;
}

/*
 * Create a listen socket and bind to all interfaces
 */

static int mca_btl_tcp_component_create_listen(void)
{
    int flags;
    struct sockaddr_in inaddr; 
    opal_socklen_t addrlen;

    /* create a listen socket for incoming connections */
    mca_btl_tcp_component.tcp_listen_sd = socket(AF_INET, SOCK_STREAM, 0);
    if(mca_btl_tcp_component.tcp_listen_sd < 0) {
        BTL_ERROR(("socket() failed with errno=%d", opal_socket_errno));
        return OMPI_ERROR;
    }
    mca_btl_tcp_set_socket_options(mca_btl_tcp_component.tcp_listen_sd);
                                                                                                      
    /* bind to all addresses and dynamically assigned port */
    memset(&inaddr, 0, sizeof(inaddr));
    inaddr.sin_family = AF_INET;
    inaddr.sin_addr.s_addr = INADDR_ANY;
    inaddr.sin_port = 0;
                                                                                                      
    if(bind(mca_btl_tcp_component.tcp_listen_sd, (struct sockaddr*)&inaddr, sizeof(inaddr)) < 0) {
        BTL_ERROR(("bind() failed with errno=%d", opal_socket_errno));
        return OMPI_ERROR;
    }
                                                                                                      
    /* resolve system assignend port */
    addrlen = sizeof(struct sockaddr_in);
    if(getsockname(mca_btl_tcp_component.tcp_listen_sd, (struct sockaddr*)&inaddr, &addrlen) < 0) {
        BTL_ERROR(("getsockname() failed with errno=%d", opal_socket_errno));
        return OMPI_ERROR;
    }
    mca_btl_tcp_component.tcp_listen_port = inaddr.sin_port;

    /* setup listen backlog to maximum allowed by kernel */
    if(listen(mca_btl_tcp_component.tcp_listen_sd, SOMAXCONN) < 0) {
        BTL_ERROR(("listen() failed with errno=%d", opal_socket_errno));
        return OMPI_ERROR;
    }

    /* set socket up to be non-blocking, otherwise accept could block */
    if((flags = fcntl(mca_btl_tcp_component.tcp_listen_sd, F_GETFL, 0)) < 0) {
        BTL_ERROR(("fcntl(F_GETFL) failed with errno=%d", opal_socket_errno));
        return OMPI_ERROR;
    } else {
        flags |= O_NONBLOCK;
        if(fcntl(mca_btl_tcp_component.tcp_listen_sd, F_SETFL, flags) < 0) {
            BTL_ERROR(("fcntl(F_SETFL) failed with errno=%d", opal_socket_errno));
            return OMPI_ERROR;
        }
    }

    /* register listen port */
    opal_event_set(
        &mca_btl_tcp_component.tcp_recv_event,
        mca_btl_tcp_component.tcp_listen_sd, 
        OPAL_EV_READ|OPAL_EV_PERSIST, 
        mca_btl_tcp_component_recv_handler, 
        0);
    opal_event_add(&mca_btl_tcp_component.tcp_recv_event,0);
    return OMPI_SUCCESS;
}

/*
 *  Register TCP module addressing information. The MCA framework
 *  will make this available to all peers. 
 */

static int mca_btl_tcp_component_exchange(void)
{
     int rc=0;
     size_t i=0;
     size_t size = mca_btl_tcp_component.tcp_num_btls * sizeof(mca_btl_tcp_addr_t);
     if(mca_btl_tcp_component.tcp_num_btls != 0) {
         mca_btl_tcp_addr_t *addrs = (mca_btl_tcp_addr_t *)malloc(size);
         for(i=0; i<mca_btl_tcp_component.tcp_num_btls; i++) {
             struct mca_btl_tcp_module_t* btl = mca_btl_tcp_component.tcp_btls[i];
             addrs[i].addr_inet    = btl->tcp_ifaddr.sin_addr;
             addrs[i].addr_port    = mca_btl_tcp_component.tcp_listen_port;
             addrs[i].addr_inuse   = 0;
         }
         rc =  mca_pml_base_modex_send(&mca_btl_tcp_component.super.btl_version, addrs, size);
         free(addrs);
     }
     return rc;
}

/*
 *  TCP module initialization:
 *  (1) read interface list from kernel and compare against module parameters
 *      then create a BTL instance for selected interfaces
 *  (2) setup TCP listen socket for incoming connection attempts
 *  (3) register BTL parameters with the MCA
 */
mca_btl_base_module_t** mca_btl_tcp_component_init(int *num_btl_modules, 
                                                   bool enable_progress_threads,
                                                   bool enable_mpi_threads)
{
    mca_btl_base_module_t **btls;
    *num_btl_modules = 0;

    /* initialize free lists */
    ompi_free_list_init( &mca_btl_tcp_component.tcp_frag_eager,
                         sizeof (mca_btl_tcp_frag_eager_t) + 
                         mca_btl_tcp_module.super.btl_eager_limit,
                         OBJ_CLASS (mca_btl_tcp_frag_eager_t),
                         mca_btl_tcp_component.tcp_free_list_num,
                         mca_btl_tcp_component.tcp_free_list_max,
                         mca_btl_tcp_component.tcp_free_list_inc,
                         NULL );
                                                                                                                                  
    ompi_free_list_init( &mca_btl_tcp_component.tcp_frag_max,
                         sizeof (mca_btl_tcp_frag_max_t) + 
                         mca_btl_tcp_module.super.btl_max_send_size,
                         OBJ_CLASS (mca_btl_tcp_frag_max_t),
                         mca_btl_tcp_component.tcp_free_list_num,
                         mca_btl_tcp_component.tcp_free_list_max,
                         mca_btl_tcp_component.tcp_free_list_inc,
                         NULL );
                                                                                                                                  
    ompi_free_list_init( &mca_btl_tcp_component.tcp_frag_user,
                         sizeof (mca_btl_tcp_frag_user_t),
                         OBJ_CLASS (mca_btl_tcp_frag_user_t),
                         mca_btl_tcp_component.tcp_free_list_num,
                         mca_btl_tcp_component.tcp_free_list_max,
                         mca_btl_tcp_component.tcp_free_list_inc,
                         NULL );
                                                                                                                                  
    /* create a BTL TCP module for selected interfaces */
    if(mca_btl_tcp_component_create_instances() != OMPI_SUCCESS)
        return 0;

    /* create a TCP listen socket for incoming connection attempts */
    if(mca_btl_tcp_component_create_listen() != OMPI_SUCCESS)
        return 0;

    /* publish TCP parameters with the MCA framework */
    if(mca_btl_tcp_component_exchange() != OMPI_SUCCESS)
        return 0;

    btls = (mca_btl_base_module_t **)malloc(mca_btl_tcp_component.tcp_num_btls * 
                  sizeof(mca_btl_base_module_t*));
    if(NULL == btls)
        return NULL;

    memcpy(btls, mca_btl_tcp_component.tcp_btls, mca_btl_tcp_component.tcp_num_btls*sizeof(mca_btl_tcp_module_t*));
    *num_btl_modules = mca_btl_tcp_component.tcp_num_btls;
    return btls;
}

/*
 *  TCP module control
 */

int mca_btl_tcp_component_control(int param, void* value, size_t size)
{
    return OMPI_SUCCESS;
}


/*
 *  Called by mca_btl_tcp_component_recv() when the TCP listen
 *  socket has pending connection requests. Accept incoming
 *  requests and queue for completion of the connection handshake.
*/


static void mca_btl_tcp_component_accept(void)
{
    while(true) {
        opal_socklen_t addrlen = sizeof(struct sockaddr_in);
        struct sockaddr_in addr;
        mca_btl_tcp_event_t *event;
        int sd = accept(mca_btl_tcp_component.tcp_listen_sd, (struct sockaddr*)&addr, &addrlen);
        if(sd < 0) {
            if(opal_socket_errno == EINTR)
                continue;
            if(opal_socket_errno != EAGAIN && opal_socket_errno != EWOULDBLOCK)
                BTL_ERROR(("accept() failed with errno %d.", opal_socket_errno));
            return;
        }
        mca_btl_tcp_set_socket_options(sd);

        /* wait for receipt of peers process identifier to complete this connection */
         
        event = OBJ_NEW(mca_btl_tcp_event_t);
        opal_event_set(&event->event, sd, OPAL_EV_READ, mca_btl_tcp_component_recv_handler, event);
        opal_event_add(&event->event, 0);
    }
}


/*
 * Event callback when there is data available on the registered 
 * socket to recv.
 */
static void mca_btl_tcp_component_recv_handler(int sd, short flags, void* user)
{
    orte_process_name_t guid;
    struct sockaddr_in addr;
    int retval;
    mca_btl_tcp_proc_t* btl_proc;
    opal_socklen_t addr_len = sizeof(addr);
    mca_btl_tcp_event_t *event = (mca_btl_tcp_event_t *)user;

    /* accept new connections on the listen socket */
    if(mca_btl_tcp_component.tcp_listen_sd == sd) {
        mca_btl_tcp_component_accept();
        return;
    }
    OBJ_RELEASE(event);

    /* recv the process identifier */
    retval = recv(sd, (char *)&guid, sizeof(guid), 0);
    if(retval != sizeof(guid)) {
        CLOSE_THE_SOCKET(sd);
        return;
    }
    ORTE_PROCESS_NAME_NTOH(guid);

    /* now set socket up to be non-blocking */
    if((flags = fcntl(sd, F_GETFL, 0)) < 0) {
        BTL_ERROR(("fcntl(F_GETFL) failed with errno=%d", opal_socket_errno));
    } else {
        flags |= O_NONBLOCK;
        if(fcntl(sd, F_SETFL, flags) < 0) {
            BTL_ERROR(("fcntl(F_SETFL) failed with errno=%d", opal_socket_errno));
        }
    }
   
    /* lookup the corresponding process */
    btl_proc = mca_btl_tcp_proc_lookup(&guid);
    if(NULL == btl_proc) {
        BTL_ERROR(("errno=%d",errno));
        CLOSE_THE_SOCKET(sd);
        return;
    }

    /* lookup peer address */
    if(getpeername(sd, (struct sockaddr*)&addr, &addr_len) != 0) {
        BTL_ERROR(("getpeername() failed with errno=%d", opal_socket_errno));
        CLOSE_THE_SOCKET(sd);
        return;
    }

    /* are there any existing peer instances will to accept this connection */
    if(mca_btl_tcp_proc_accept(btl_proc, &addr, sd) == false) {
        CLOSE_THE_SOCKET(sd);
        return;
    }
}