smpd_state_machine.c

fortran并行计算包

    h[1] = smpd_process.hCloseStdinThreadEvent;
    for (;;)
    {
	/*smpd_dbg_printf("waiting for input from stdin\n");*/
	result = WaitForMultipleObjects(2, h, FALSE, INFINITE);
	if (result == WAIT_OBJECT_0)
	{
	    if (fgets(str, SMPD_MAX_CMD_LENGTH, stdin))
	    {
		len = (DWORD)strlen(str);
		smpd_dbg_printf("forwarding stdin: '%s'\n", str);
		if (send(hWrite, str, len, 0) == SOCKET_ERROR)
		{
		    smpd_err_printf("unable to forward stdin, send failed, error %d\n", WSAGetLastError());
		    return;
		}
	    }
	    else
	    {
		/* fgets failed, what do I do? */
		if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
		{
		    smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		}
		if (closesocket(hWrite) == SOCKET_ERROR)
		{
		    smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
		}
		smpd_dbg_printf("fgets failed, closing stdin reader thread.\n");
		return;
	    }
	}
	else if (result == WAIT_OBJECT_0 + 1)
	{
	    if (shutdown(hWrite, SD_BOTH) == SOCKET_ERROR)
	    {
		smpd_err_printf("shutdown failed, sock %d, error %d\n", hWrite, WSAGetLastError());
	    }
	    if (closesocket(hWrite) == SOCKET_ERROR)
	    {
		smpd_err_printf("closesocket failed, sock %d, error %d\n", hWrite, WSAGetLastError());
	    }
	    smpd_dbg_printf("hCloseStdinThreadEvent signalled, closing stdin reader thread.\n");
	    return;
	}
	else
	{
	    smpd_err_printf("stdin wait failed, error %d\n", GetLastError());
	    return;
	}
    }
}
#endif
#else
#ifdef USE_PTHREAD_STDIN_REDIRECTION
void *smpd_pthread_stdin_thread(void *p)
{
    int read_fd, write_fd;
    char ch;
    size_t num_read;
    fd_set set;
    int n;
    /*struct timeval tv;*/

    write_fd = smpd_process.stdin_write;
    read_fd = fileno(stdin);

    for(;;)
    {
	FD_ZERO(&set);
	FD_SET(read_fd, &set);
	FD_SET(write_fd, &set);

	/*
	  tv.tv_sec = -1;
	  tv.tv_usec = -1;
	*/
	n = select(write_fd+1, &set, NULL, NULL, NULL/*&tv*/);
	if (n<0)
	{
	    if (errno = EINTR || errno == 0)
		continue;
	    goto fn_exit;
	}

	if (FD_ISSET(write_fd, &set))
	{
	    /* A byte is sent on the write_fd to inform the thread to exit */
	    read(write_fd, &ch, 1);
	    goto fn_exit;
	}
	if (FD_ISSET(read_fd, &set))
	{
	    num_read = read(read_fd, &ch, 1);
	    if (num_read > 0)
	    {
		if (write(write_fd, &ch, 1) != 1)
		{
		    smpd_dbg_printf("stdin redirection write failed, error %d\n", errno);
		    goto fn_exit;
		}
	    }
	    else if (num_read == 0)
	    {
		/* Read thread closed, should we exit? */
		goto fn_exit;
	    }
	    else if (num_read == -1)
	    {
		/* An error occurred while reading */
		if (errno != EINTR && errno != 0)
		{
		    smpd_err_printf("reading from stdin failed with error %d\n", errno);
		    goto fn_exit;
		}
	    }
	}
    }
fn_exit:
    close(write_fd);
    return NULL;
}

#undef FCNAME
#define FCNAME "smpd_cancel_stdin_thread"
int smpd_cancel_stdin_thread()
{
    char ch = 0;
    smpd_enter_fn(FCNAME);
    /*printf("cancelling stin redirection thread\n");*/
    if (smpd_process.stdin_thread == NULL)
    {
	smpd_exit_fn(FCNAME);
	return SMPD_SUCCESS;
    }
    write(smpd_process.stdin_read, &ch, 1);
    pthread_join(smpd_process.stdin_thread, NULL);
    /*pthread_cancel(smpd_process.stdin_thread);*/
    smpd_process.stdin_thread = NULL;
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#endif
#endif

#undef FCNAME
#define FCNAME "smpd_get_state_string"
const char * smpd_get_state_string(smpd_state_t state)
{
    static char unknown_str[100];
    const char *result;

    smpd_enter_fn(FCNAME);
    switch (state)
    {
    case SMPD_IDLE:                            result = "SMPD_IDLE";                            break;
    case SMPD_MPIEXEC_CONNECTING_TREE:         result = "SMPD_MPIEXEC_CONNECTING_TREE";         break;
    case SMPD_MPIEXEC_CONNECTING_SMPD:         result = "SMPD_MPIEXEC_CONNECTING_SMPD";         break;
    case SMPD_CONNECTING:                      result = "SMPD_CONNECTING";                      break;
    case SMPD_READING_CHALLENGE_STRING:        result = "SMPD_READING_CHALLENGE_STRING";        break;
    case SMPD_WRITING_CHALLENGE_RESPONSE:      result = "SMPD_WRITING_CHALLENGE_RESPONSE";      break;
    case SMPD_READING_CONNECT_RESULT:          result = "SMPD_READING_CONNECT_RESULT";          break;
    case SMPD_WRITING_CHALLENGE_STRING:        result = "SMPD_WRITING_CHALLENGE_STRING";        break;
    case SMPD_READING_CHALLENGE_RESPONSE:      result = "SMPD_READING_CHALLENGE_RESPONSE";      break;
    case SMPD_WRITING_CONNECT_RESULT:          result = "SMPD_WRITING_CONNECT_RESULT";          break;
    case SMPD_READING_STDIN:                   result = "SMPD_READING_STDIN";                   break;
    case SMPD_WRITING_DATA_TO_STDIN:           result = "SMPD_WRITING_DATA_TO_STDIN";           break;
    case SMPD_READING_STDOUT:                  result = "SMPD_READING_STDOUT";                  break;
    case SMPD_READING_STDERR:                  result = "SMPD_READING_STDERR";                  break;
    case SMPD_READING_CMD_HEADER:              result = "SMPD_READING_CMD_HEADER";              break;
    case SMPD_READING_CMD:                     result = "SMPD_READING_CMD";                     break;
    case SMPD_WRITING_CMD:                     result = "SMPD_WRITING_CMD";                     break;
    case SMPD_SMPD_LISTENING:                  result = "SMPD_SMPD_LISTENING";                  break;
    case SMPD_MGR_LISTENING:                   result = "SMPD_MGR_LISTENING";                   break;
    case SMPD_PMI_LISTENING: 	               result = "SMPD_PMI_LISTENING";                   break;
    case SMPD_SINGLETON_CLIENT_LISTENING:      result = "SMPD_SINGLETON_CLIENT_LISTENING";      break;
    case SMPD_SINGLETON_MPIEXEC_CONNECTING:    result = "SMPD_SINGLETON_MPIEXEC_CONNECTING";    break;
    case SMPD_SINGLETON_READING_PMI_INFO:      result = "SMPD_SINGLETON_READING_PMI_INFO";      break;
    case SMPD_SINGLETON_WRITING_PMI_INFO:      result = "SMPD_SINGLETON_WRITING_PMI_INFO";      break;
    case SMPD_SINGLETON_DONE:                  result = "SMPD_SINGLETON_DONE";                  break;
    case SMPD_PMI_SERVER_LISTENING:            result = "SMPD_PMI_SERVER_LISTENING";            break;
    case SMPD_READING_SESSION_REQUEST:         result = "SMPD_READING_SESSION_REQUEST";         break;
    case SMPD_WRITING_SMPD_SESSION_REQUEST:    result = "SMPD_WRITING_SMPD_SESSION_REQUEST";    break;
    case SMPD_WRITING_PROCESS_SESSION_REQUEST: result = "SMPD_WRITING_PROCESS_SESSION_REQUEST"; break;
    case SMPD_WRITING_PMI_SESSION_REQUEST:     result = "SMPD_WRITING_PMI_SESSION_REQUEST";     break;
    case SMPD_WRITING_PWD_REQUEST:             result = "SMPD_WRITING_PWD_REQUEST";             break;
    case SMPD_WRITING_NO_PWD_REQUEST:          result = "SMPD_WRITING_NO_PWD_REQUEST";          break;
    case SMPD_READING_PWD_REQUEST:             result = "SMPD_READING_PWD_REQUEST";             break;
    case SMPD_READING_SMPD_PASSWORD:           result = "SMPD_READING_SMPD_PASSWORD";           break;
    case SMPD_WRITING_CRED_REQUEST:            result = "SMPD_WRITING_CRED_REQUEST";            break;
    case SMPD_READING_CRED_ACK:                result = "SMPD_READING_CRED_ACK";                break;
    case SMPD_WRITING_CRED_ACK_YES:            result = "SMPD_WRITING_CRED_ACK_YES";            break;
    case SMPD_WRITING_CRED_ACK_NO:             result = "SMPD_WRITING_CRED_ACK_NO";             break;
    case SMPD_READING_ACCOUNT:                 result = "SMPD_READING_ACCOUNT";                 break;
    case SMPD_READING_PASSWORD:                result = "SMPD_READING_PASSWORD";                break;
    case SMPD_WRITING_ACCOUNT:                 result = "SMPD_WRITING_ACCOUNT";                 break;
    case SMPD_WRITING_PASSWORD:                result = "SMPD_WRITING_PASSWORD";                break;
    case SMPD_WRITING_NO_CRED_REQUEST:         result = "SMPD_WRITING_NO_CRED_REQUEST";         break;
    case SMPD_READING_CRED_REQUEST:            result = "SMPD_READING_CRED_REQUEST";            break;
    case SMPD_WRITING_RECONNECT_REQUEST:       result = "SMPD_WRITING_RECONNECT_REQUEST";       break;
    case SMPD_WRITING_NO_RECONNECT_REQUEST:    result = "SMPD_WRITING_NO_RECONNECT_REQUEST";    break;
    case SMPD_READING_RECONNECT_REQUEST:       result = "SMPD_READING_RECONNECT_REQUEST";       break;
    case SMPD_READING_SESSION_HEADER:          result = "SMPD_READING_SESSION_HEADER";          break;
    case SMPD_WRITING_SESSION_HEADER:          result = "SMPD_WRITING_SESSION_HEADER";          break;
    case SMPD_READING_SMPD_RESULT:             result = "SMPD_READING_SMPD_RESULT";             break;
    case SMPD_READING_PROCESS_RESULT:          result = "SMPD_READING_PROCESS_RESULT";          break;
    case SMPD_WRITING_SESSION_ACCEPT:          result = "SMPD_WRITING_SESSION_ACCEPT";          break;
    case SMPD_WRITING_SESSION_REJECT:          result = "SMPD_WRITING_SESSION_REJECT";          break;
    case SMPD_WRITING_PROCESS_SESSION_ACCEPT:  result = "SMPD_WRITING_PROCESS_SESSION_REJECT";  break;
    case SMPD_WRITING_PROCESS_SESSION_REJECT:  result = "SMPD_WRITING_PROCESS_SESSION_REJECT";  break;
    case SMPD_RECONNECTING:                    result = "SMPD_RECONNECTING";                    break;
    case SMPD_EXITING:                         result = "SMPD_EXITING";                         break;
    case SMPD_CLOSING:                         result = "SMPD_CLOSING";                         break;
    case SMPD_WRITING_SMPD_PASSWORD:           result = "SMPD_WRITING_SMPD_PASSWORD";           break;
    case SMPD_READING_SSPI_HEADER:             result = "SMPD_READING_SSPI_HEADER";             break;
    case SMPD_READING_SSPI_BUFFER:             result = "SMPD_READING_SSPI_BUFFER";             break;
    case SMPD_WRITING_SSPI_HEADER:             result = "SMPD_WRITING_SSPI_HEADER";             break;
    case SMPD_WRITING_SSPI_BUFFER:             result = "SMPD_WRITING_SSPI_BUFFER";             break;
    case SMPD_WRITING_DELEGATE_REQUEST:        result = "SMPD_WRITING_DELEGATE_REQUEST";        break;
    case SMPD_READING_DELEGATE_REQUEST_RESULT: result = "SMPD_READING_DELEGATE_REQUEST_RESULT"; break;
    case SMPD_WRITING_IMPERSONATE_RESULT:      result = "SMPD_WRITING_IMPERSONATE_RESULT";      break;
    case SMPD_WRITING_CRED_ACK_SSPI:           result = "SMPD_WRITING_CRED_ACK_SSPI";           break;
    case SMPD_READING_CLIENT_SSPI_HEADER:      result = "SMPD_READING_CLIENT_SSPI_HEADER";      break;
    case SMPD_READING_CLIENT_SSPI_BUFFER:      result = "SMPD_READING_CLIENT_SSPI_BUFFER";      break;
    case SMPD_WRITING_CLIENT_SSPI_HEADER:      result = "SMPD_WRITING_CLIENT_SSPI_HEADER";      break;
    case SMPD_WRITING_CLIENT_SSPI_BUFFER:      result = "SMPD_WRITING_CLIENT_SSPI_BUFFER";      break;
    case SMPD_READING_TIMEOUT:                 result = "SMPD_READING_TIMEOUT";                 break;
    case SMPD_READING_MPIEXEC_ABORT:           result = "SMPD_READING_MPIEXEC_ABORT";           break;
    case SMPD_RESTARTING:                      result = "SMPD_RESTARTING";                      break;
    case SMPD_DONE:                            result = "SMPD_DONE";                            break;
    case SMPD_CONNECTING_RPMI:                 result = "SMPD_CONNECTING_RPMI";                 break;
    case SMPD_CONNECTING_PMI:                  result = "SMPD_CONNECTING_PMI";                  break;
    case SMPD_WRITING_SSPI_REQUEST:            result = "SMPD_WRITING_SSPI_REQUEST";            break;
    case SMPD_READING_PMI_ID:                  result = "SMPD_READING_PMI_ID";                  break;
    case SMPD_WRITING_PMI_ID:                  result = "SMPD_WRITING_PMI_ID";                  break;
    case SMPD_WRITING_DELEGATE_REQUEST_RESULT: result = "SMPD_WRITING_DELEGATE_REQUEST_RESULT"; break;
    case SMPD_READING_IMPERSONATE_RESULT:      result = "SMPD_READING_IMPERSONATE_RESULT";      break;
    case SMPD_WRITING_CRED_ACK_SSPI_JOB_KEY:   result = "SMPD_WRITING_CRED_ACK_SSPI_JOB_KEY";   break;
    case SMPD_WRITING_SSPI_JOB_KEY:            result = "SMPD_WRITING_SSPI_JOB_KEY";            break;
    case SMPD_READING_SSPI_JOB_KEY:            result = "SMPD_READING_SSPI_JOB_KEY";            break;
    default:
	sprintf(unknown_str, "unknown state %d", state);
	result = unknown_str;
	break;
    }
    smpd_exit_fn(FCNAME);
    return result;
}

#undef FCNAME
#define FCNAME "smpd_verify_version"
SMPD_BOOL smpd_verify_version(const char *challenge)
{
    char version[100];
    char *space_char;

    /* The first part of the challenge string up to the first space is the smpd version */

    smpd_enter_fn(FCNAME);
    space_char = strchr(challenge, ' ');
    if (space_char != NULL && space_char - challenge < 100)
    {
	strncpy(version, challenge, space_char - challenge);
	version[space_char - challenge] = '\0';
	if (strcmp(version, SMPD_VERSION) == 0)
	{
	    smpd_exit_fn(FCNAME);
	    return SMPD_TRUE;
	}
    }

    smpd_exit_fn(FCNAME);
    return SMPD_FALSE;
}

#undef FCNAME
#define FCNAME "smpd_state_reading_challenge_string"
int smpd_state_reading_challenge_string(smpd_context_t *context, MPIDU_Sock_event_t *event_ptr)
{
    int result;
    char phrase[SMPD_PASSPHRASE_MAX_LENGTH];

    smpd_enter_fn(FCNAME);

    if (event_ptr->error != MPI_SUCCESS)
    {
	smpd_err_printf("unable to read the challenge string, %s.\n", get_sock_error_string(event_ptr->error));
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_dbg_printf("read challenge string: '%s'\n", context->pszChallengeResponse);
    context->read_state = SMPD_IDLE;

    /* check to see if we are simply trying to get the version */
    if (smpd_process.do_console == 1 && smpd_process.builtin_cmd == SMPD_CMD_VERSION)
    {
	strtok(context->pszChallengeResponse, " ");
	printf("%s\n", context->pszChallengeResponse);
	fflush(stdout);
	smpd_exit(0);
    }

    if (smpd_verify_version(context->pszChallengeResponse) == SMPD_TRUE)
    {
	strcpy(phrase, smpd_process.passphrase);
	/* crypt the passphrase + the challenge */
	if (strlen(phrase) + strlen(context->pszChallengeResponse) > SMPD_PASSPHRASE_MAX_LENGTH)
	{
	    smpd_err_printf("smpd_client_authenticate: unable to process passphrase - too long.\n");
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
	strcat(phrase, context->pszChallengeResponse);

	/*smpd_dbg_printf("crypting: %s\n", phrase);*/
	smpd_hash(phrase, (int)strlen(phrase), context->pszChallengeResponse, SMPD_AUTHENTICATION_STR_LEN);
	/* save the response to be used as a prefix to the password when encrypting the password */
	MPIU_Strncpy(smpd_process.encrypt_prefix, context->pszChallengeResponse, SMPD_MAX_PASSWORD_LENGTH);
    }
    else
    {
	strcpy(context->pszChallengeResponse, SMPD_VERSION_FAILURE);
    }

    /* write the response */
    /*smpd_dbg_printf("writing response: %s\n", pszStr);*/
    context->write_state = SMPD_WRITING_CHALLENGE_RESPONSE;
    result = MPIDU_Sock_post_write(context->sock, context->pszChallengeResponse, SMPD_AUTHENTICATION_STR_LEN, SMPD_AUTHENTICATION_STR_LEN, NULL);
    if (result != MPI_SUCCESS)
    {
	smpd_err_printf("unable to post a write of the encrypted response string,\nsock error: %s\n",
	    get_sock_error_string(result));
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_state_writing_challenge_response"
int smpd_state_writing_challenge_response(smpd_context_t *context, MPIDU_Sock_event_t *event_ptr)
{
    int result;

    smpd_enter_fn(FCNAME);
    if (event_ptr->error != MPI_SUCCESS)
    {
	smpd_err_printf("unable to write the challenge response, %s.\n", get_sock_error_string(event_ptr->error));
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_dbg_printf("wrote challenge response: '%s'\n", context->pszChallengeResponse);
    context->write_state = SMPD_IDLE;
    context->read_state = SMPD_READING_CONNECT_RESULT;
    result = MPIDU_Sock_post_read(context->sock, context->pszChallengeResponse, SMPD_AUTHENTICATION_STR_LEN, SMPD_AUTHENTICATION_STR_LEN, NULL);
    if (result != MPI_SUCCESS)
    {
	smpd_err_printf("unable to post a read for the connect response,\nsock error: %s\n",
	    get_sock_error_string(result));
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_state_reading_connect_result"
int smpd_state_reading_connect_result(smpd_context_t *context, MPIDU_Sock_event_t *event_ptr)
{
    int result;

    smpd_enter_fn(FCNAME);
    if (event_ptr->error != MPI_SUCCESS)
    {