src/mca/oob/tcp/oob_tcp_listener.c
/*
 * The listen thread created when listen_mode is threaded.  Accepts
 * incoming connections and places them in a queue for further
 * processing
 *
 * Runs until prte_oob_tcp_compnent.shutdown is set to true.
 */
The listen thread created when listen_mode is threaded.  Accepts
incoming connections and places them in a queue for further
processing

Runs until prte_oob_tcp_compnent.shutdown is set to true.
The listen thread created when listen_mode is threaded . Accepts incoming connections and places them in a queue for further processing Runs until prte_oob_tcp_compnent . shutdown is set to true .
static void *listen_thread(prte_object_t *obj)
{
    int rc, max, accepted_connections, sd;
    prte_socklen_t addrlen = sizeof(struct sockaddr_storage);
    prte_oob_tcp_pending_connection_t *pending_connection;
    struct timeval timeout;
    fd_set readfds;
    prte_oob_tcp_listener_t *listener;
    while (prte_oob_tcp_component.listen_thread_active) {
        FD_ZERO(&readfds);
        max = -1;
        PRTE_LIST_FOREACH(listener, &prte_oob_tcp_component.listeners, prte_oob_tcp_listener_t)
        {
            FD_SET(listener->sd, &readfds);
            max = (listener->sd > max) ? listener->sd : max;
        }
        FD_SET(prte_oob_tcp_component.stop_thread[0], &readfds);
        max = (prte_oob_tcp_component.stop_thread[0] > max) ? prte_oob_tcp_component.stop_thread[0]
                                                            : max;
        timeout.tv_sec = prte_oob_tcp_component.listen_thread_tv.tv_sec;
        timeout.tv_usec = prte_oob_tcp_component.listen_thread_tv.tv_usec;
        rc = select(max + 1, &readfds, NULL, NULL, &timeout);
        if (!prte_oob_tcp_component.listen_thread_active) {
            return NULL;
        }
        if (rc < 0) {
            if (EAGAIN != prte_socket_errno && EINTR != prte_socket_errno) {
                perror("select");
            }
            continue;
        }
        do {
            accepted_connections = 0;
            PRTE_LIST_FOREACH(listener, &prte_oob_tcp_component.listeners, prte_oob_tcp_listener_t)
            {
                sd = listener->sd;
                if (0 == FD_ISSET(sd, &readfds)) {
                    continue;
                }
                pending_connection = PRTE_NEW(prte_oob_tcp_pending_connection_t);
                prte_event_set(prte_event_base, &pending_connection->ev, -1, PRTE_EV_WRITE,
                               connection_handler, pending_connection);
                prte_event_set_priority(&pending_connection->ev, PRTE_MSG_PRI);
                pending_connection->fd = accept(sd, (struct sockaddr *) &(pending_connection->addr),
                                                &addrlen);
                if (pending_connection->fd < 0) {
                    PRTE_RELEASE(pending_connection);
                    if (EAGAIN == prte_socket_errno || EWOULDBLOCK == prte_socket_errno) {
                        continue;
                    }
                    else if (EMFILE == prte_socket_errno) {
                        CLOSE_THE_SOCKET(sd);
                        PRTE_ERROR_LOG(PRTE_ERR_SYS_LIMITS_SOCKETS);
                        prte_show_help("help-oob-tcp.txt", "accept failed", true,
                                       prte_process_info.nodename, prte_socket_errno,
                                       strerror(prte_socket_errno), "Out of file descriptors");
                        goto done;
                    }
                    else {
                        prte_show_help("help-oob-tcp.txt", "accept failed", true,
                                       prte_process_info.nodename, prte_socket_errno,
                                       strerror(prte_socket_errno),
                                       "Unknown cause; job will try to continue");
                        continue;
                    }
                }
                prte_output_verbose(OOB_TCP_DEBUG_CONNECT, prte_oob_base_framework.framework_output,
                                    "%s prte_oob_tcp_listen_thread: incoming connection: "
                                    "(%d, %d) %s:%d\n",
                                    PRTE_NAME_PRINT(PRTE_PROC_MY_NAME), pending_connection->fd,
                                    prte_socket_errno,
                                    prte_net_get_hostname(
                                        (struct sockaddr *) &pending_connection->addr),
                                    prte_net_get_port(
                                        (struct sockaddr *) &pending_connection->addr));
                if (1024 >= listener->port) {
                    uint16_t inport;
                    inport = prte_net_get_port((struct sockaddr *) &pending_connection->addr);
                    if (1024 < inport) {
                        prte_show_help("help-oob-tcp.txt", "privilege failure", true,
                                       prte_process_info.nodename, listener->port,
                                       prte_net_get_hostname(
                                           (struct sockaddr *) &pending_connection->addr),
                                       inport);
                        CLOSE_THE_SOCKET(pending_connection->fd);
                        PRTE_RELEASE(pending_connection);
                        continue;
                    }
                }
                PRTE_POST_OBJECT(pending_connection);
                prte_event_active(&pending_connection->ev, PRTE_EV_WRITE, 1);
                accepted_connections++;
            }
        } while (accepted_connections > 0);
    }
done:
#if 0
    prte_output_verbose(OOB_TCP_DEBUG_CONNECT, prte_oob_base_framework.framework_output,
                        "%s prte_oob_tcp_listen_thread: switching to event lib",
                        PRTE_NAME_PRINT(PRTE_PROC_MY_NAME));
    PRTE_LIST_FOREACH(listener, &prte_oob_tcp_component.listeners, prte_oob_tcp_listener_t) {
        prte_event_set(prte_event_base, listener->event,
                   listener->sd,
                   PRTE_EV_READ|PRTE_EV_PERSIST,
                   connection_event_handler,
                   0);
        prte_event_set_priority(listener->event, PRTE_MSG_PRI);
        prte_event_add(listener->event, 0);
    }
#endif
    return NULL;
}
static void * listen_thread ( prte_object_t * obj ) { int rc , max , accepted_connections , sd ; prte_socklen_t addrlen = sizeof ( struct sockaddr_storage ) ; prte_oob_tcp_pending_connection_t * pending_connection ; struct timeval timeout ; fd_set readfds ; prte_oob_tcp_listener_t * listener ; while ( prte_oob_tcp_component . listen_thread_active ) { FD_ZERO ( & readfds ) ; max = -1 ; PRTE_LIST_FOREACH ( listener , & prte_oob_tcp_component . listeners , prte_oob_tcp_listener_t ) { FD_SET ( listener -> sd , & readfds ) ; max = ( listener -> sd > max ) ? listener -> sd : max ; } FD_SET ( prte_oob_tcp_component . stop_thread [ 0 ] , & readfds ) ; max = ( prte_oob_tcp_component . stop_thread [ 0 ] > max ) ? prte_oob_tcp_component . stop_thread [ 0 ] : max ; timeout . tv_sec = prte_oob_tcp_component . listen_thread_tv . tv_sec ; timeout . tv_usec = prte_oob_tcp_component . listen_thread_tv . tv_usec ; rc = select ( max + 1 , & readfds , NULL , NULL , & timeout ) ; if ( ! prte_oob_tcp_component . listen_thread_active ) { return NULL ; } if ( rc < 0 ) { if ( EAGAIN != prte_socket_errno && EINTR != prte_socket_errno ) { perror ( " " ) ; } continue ; } do { accepted_connections = 0 ; PRTE_LIST_FOREACH ( listener , & prte_oob_tcp_component . listeners , prte_oob_tcp_listener_t ) { sd = listener -> sd ; if ( 0 == FD_ISSET ( sd , & readfds ) ) { continue ; } pending_connection = PRTE_NEW ( prte_oob_tcp_pending_connection_t ) ; prte_event_set ( prte_event_base , & pending_connection -> ev , -1 , PRTE_EV_WRITE , connection_handler , pending_connection ) ; prte_event_set_priority ( & pending_connection -> ev , PRTE_MSG_PRI ) ; pending_connection -> fd = accept ( sd , ( struct sockaddr * ) & ( pending_connection -> addr ) , & addrlen ) ; if ( pending_connection -> fd < 0 ) { PRTE_RELEASE ( pending_connection ) ; if ( EAGAIN == prte_socket_errno || EWOULDBLOCK == prte_socket_errno ) { continue ; } else if ( EMFILE == prte_socket_errno ) { CLOSE_THE_SOCKET ( sd ) ; PRTE_ERROR_LOG ( PRTE_ERR_SYS_LIMITS_SOCKETS ) ; prte_show_help ( " " , " " , true , prte_process_info . nodename , prte_socket_errno , strerror ( prte_socket_errno ) , " " ) ; goto done ; } else { prte_show_help ( " " , " " , true , prte_process_info . nodename , prte_socket_errno , strerror ( prte_socket_errno ) , " " ) ; continue ; } } prte_output_verbose ( OOB_TCP_DEBUG_CONNECT , prte_oob_base_framework . framework_output , " " " \n " , PRTE_NAME_PRINT ( PRTE_PROC_MY_NAME ) , pending_connection -> fd , prte_socket_errno , prte_net_get_hostname ( ( struct sockaddr * ) & pending_connection -> addr ) , prte_net_get_port ( ( struct sockaddr * ) & pending_connection -> addr ) ) ; if ( 1024 >= listener -> port ) { uint16_t inport ; inport = prte_net_get_port ( ( struct sockaddr * ) & pending_connection -> addr ) ; if ( 1024 < inport ) { prte_show_help ( " " , " " , true , prte_process_info . nodename , listener -> port , prte_net_get_hostname ( ( struct sockaddr * ) & pending_connection -> addr ) , inport ) ; CLOSE_THE_SOCKET ( pending_connection -> fd ) ; PRTE_RELEASE ( pending_connection ) ; continue ; } } PRTE_POST_OBJECT ( pending_connection ) ; prte_event_active ( & pending_connection -> ev , PRTE_EV_WRITE , 1 ) ; accepted_connections ++ ; } } while ( accepted_connections > 0 ) ; } done : #if 0 prte_output_verbose ( OOB_TCP_DEBUG_CONNECT , prte_oob_base_framework . framework_output , " " , PRTE_NAME_PRINT ( PRTE_PROC_MY_NAME ) ) ; PRTE_LIST_FOREACH ( listener , & prte_oob_tcp_component . listeners , prte_oob_tcp_listener_t ) { prte_event_set ( prte_event_base , listener -> event , listener -> sd , PRTE_EV_READ | PRTE_EV_PERSIST , connection_event_handler , 0 ) ; prte_event_set_priority ( listener -> event , PRTE_MSG_PRI ) ; prte_event_add ( listener -> event , 0 ) ; } #endif return NULL ; }
The listen thread created when listen_mode is threaded.
The listen thread created when listen_mode is threaded .
/* only execute during the initial VM startup stage - once
     * all the initial daemons have reported in, we will revert
     * to the event method for handling any further connections
     * so as to minimize overhead
     */ /* add the stop_thread fd */ /* set timeout interval */ /* Block in a select to avoid hammering the cpu.  If a connection
         * comes in, we'll get woken up right away.
         */ /* we've been asked to terminate */ /* Spin accepting connections until all active listen sockets
         * do not have any incoming connections, pushing each connection
         * onto the event queue for processing
         */ /* according to the man pages, select replaces the given descriptor
                 * set with a subset consisting of those descriptors that are ready
                 * for the specified operation - in this case, a read. So we need to
                 * first check to see if this file descriptor is included in the
                 * returned subset
                 */ /* this descriptor is not included */ /* this descriptor is ready to be read, which means a connection
                 * request has been received - so harvest it. All we want to do
                 * here is accept the connection and push the info onto the event
                 * library for subsequent processing - we don't want to actually
                 * process the connection here as it takes too long, and so the
                 * OS might start rejecting connections due to timeout.
                 */ /* check for < 0 as indicating an error upon accept */ /* Non-fatal errors */ /* If we run out of file descriptors, log an extra
                       warning (so that the user can know to fix this
                       problem) and abandon all hope. */ /* For all other cases, print a
                       warning but try to continue */ /* if we are on a privileged port, we only accept connections
                 * from other privileged sockets. A privileged port is one
                 * whose port is less than 1024 on Linux, so we'll check for that. */ /* someone tried to cross-connect privileges,
                         * say something */ /* activate the event */ /* once we complete the initial launch, the "flood" of connections
     * will end - only connection requests from local procs, connect/accept
     * operations across mpirun instances, or the occasional tool will need
     * to be serviced. As these are relatively small events, we can easily
     * handle them in the context of the event library and no longer require
     * a separate connection harvesting thread. So switch over to the event
     * lib handler now
     */ /* setup to listen via event library */