smpd_connect.c

fortran并行计算包

	if (closesocket(sock) == SOCKET_ERROR)
	{
	    smpd_err_printf("closesocket failed, sock %d, error %d\n", sock, WSAGetLastError());
	}
	*pRead = INVALID_SOCKET;
	*pWrite = INVALID_SOCKET;
	smpd_exit_fn(FCNAME);
	return error;
    }

    /* set the nodelay option */
    b = TRUE;
    setsockopt(*pWrite, IPPROTO_TCP, TCP_NODELAY, (char*)&b, sizeof(BOOL));

    /* Set the linger on close option */
    /*
    linger.l_onoff = 1 ;
    linger.l_linger = 60;
    setsockopt(*pWrite, SOL_SOCKET, SO_LINGER, (char*)&linger, sizeof(linger));
    */

    sockAddr.sin_addr.s_addr = inet_addr("127.0.0.1");

    /* connect to myself */
    if (connect(*pWrite, (SOCKADDR*)&sockAddr, sizeof(sockAddr)) == SOCKET_ERROR)
    {
	error = WSAGetLastError();
	if (closesocket(*pWrite) == SOCKET_ERROR)
	{
	    smpd_err_printf("closesocket failed, sock %d, error %d\n", *pWrite, WSAGetLastError());
	}
	if (closesocket(sock) == SOCKET_ERROR)
	{
	    smpd_err_printf("closesocket failed, sock %d, error %d\n", sock, WSAGetLastError());
	}
	*pRead = INVALID_SOCKET;
	*pWrite = INVALID_SOCKET;
	smpd_exit_fn(FCNAME);
	return error;
    }

    /* Accept the connection from myself */
    len = sizeof(sockAddr);
    *pRead = accept(sock, (SOCKADDR*)&sockAddr, &len);

    /* set the nodelay option */
    b = TRUE;
    setsockopt(*pRead, IPPROTO_TCP, TCP_NODELAY, (char*)&b, sizeof(BOOL));

    if (closesocket(sock) == SOCKET_ERROR)
    {
	smpd_err_printf("closesocket failed, sock %d, error %d\n", sock, WSAGetLastError());
    }
    smpd_exit_fn(FCNAME);
    return 0;
}
#endif

#undef FCNAME
#define FCNAME "smpd_init_process"
int smpd_init_process(void)
{
#ifdef HAVE_WINDOWS_H
    HMODULE hModule;
#else
    char *homedir;
    struct stat s;
#endif
#ifdef HAVE_SIGACTION
    struct sigaction act;
#endif

    smpd_enter_fn(FCNAME);

    /* initialize the debugging output print engine */
    smpd_init_printf();

    /* tree data */
    smpd_process.parent_id = -1;
    smpd_process.id = -1;
    smpd_process.level = -1;
    smpd_process.left_context = NULL;
    smpd_process.right_context = NULL;
    smpd_process.parent_context = NULL;
    smpd_process.set = MPIDU_SOCK_INVALID_SET;

    /* local data */
#ifdef HAVE_WINDOWS_H
    hModule = GetModuleHandle(NULL);
    if (!GetModuleFileName(hModule, smpd_process.pszExe, SMPD_MAX_EXE_LENGTH)) 
	smpd_process.pszExe[0] = '\0';
#else
    smpd_process.pszExe[0] = '\0';
#endif
    strcpy(smpd_process.SMPDPassword, SMPD_DEFAULT_PASSWORD);
    smpd_process.bPasswordProtect = SMPD_FALSE;
    smpd_process.bService = SMPD_FALSE;
    smpd_get_hostname(smpd_process.host, SMPD_MAX_HOST_LENGTH);
    smpd_process.UserAccount[0] = '\0';
    smpd_process.UserPassword[0] = '\0';
    smpd_process.closing = SMPD_FALSE;
    smpd_process.root_smpd = SMPD_FALSE;

    srand(smpd_getpid());

#ifdef HAVE_SIGACTION
    memset(&act, 0, sizeof(act));
    act.sa_handler = smpd_child_handler;
#ifdef SA_NODEFER
    act.sa_flags = SA_NOCLDSTOP | SA_NODEFER;
#else
    act.sa_flags = SA_NOCLDSTOP;
#endif
    sigaction(SIGCHLD, &act, NULL);
#endif

#ifdef HAVE_WINDOWS_H
    smpd_process.hBombDiffuseEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
    smpd_process.hLaunchProcessMutex = CreateMutex(NULL, FALSE, NULL);
#else
    /* Setting the smpd_filename to default filename is done in smpd_get_smpd_data()
     * Avoid duplicating the effort
     */ 
	/*
    homedir = getenv("HOME");
    if(homedir != NULL){
        strcpy(smpd_process.smpd_filename, homedir);
        if (smpd_process.smpd_filename[strlen(smpd_process.smpd_filename)-1] != '/')
	    strcat(smpd_process.smpd_filename, "/.smpd");
        else
	    strcat(smpd_process.smpd_filename, ".smpd");
    }else{
	strcpy(smpd_process.smpd_filename, ".smpd");
    }
    if (stat(smpd_process.smpd_filename, &s) == 0)
    {
	if (s.st_mode & 00077)
	{
	    printf("smpd file, %s, cannot be readable by anyone other than the current user, please set the permissions accordingly (0600).\n", smpd_process.smpd_filename);
	    smpd_process.smpd_filename[0] = '\0';
	}
    }
    else
    {
	smpd_process.smpd_filename[0] = '\0';
    }
	*/
#endif
	smpd_process.smpd_filename[0] = '\0';
	smpd_process.passphrase[0] = '\0';
    /* smpd_init_process() should not try to get the passphrase. Just initialize the values */
    /* smpd_get_smpd_data("phrase", smpd_process.passphrase, SMPD_PASSPHRASE_MAX_LENGTH); */

    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_init_context"
int smpd_init_context(smpd_context_t *context, smpd_context_type_t type, MPIDU_Sock_set_t set, MPIDU_Sock_t sock, int id)
{
    int result;

    smpd_enter_fn(FCNAME);
    context->type = type;
    context->target = SMPD_TARGET_UNDETERMINED;
    context->access = SMPD_ACCESS_USER_PROCESS;/*SMPD_ACCESS_NONE;*/
    context->host[0] = '\0';
    context->id = id;
    context->rank = 0;
    context->write_list = NULL;
    context->wait_list = NULL;
    smpd_init_command(&context->read_cmd);
    context->next = NULL;
    context->set = set;
    context->sock = sock;
    context->state = SMPD_IDLE;
    context->read_state = SMPD_IDLE;
    context->write_state = SMPD_IDLE;
    context->account[0] = '\0';
    context->domain[0] = '\0';
    context->full_domain[0] = '\0';
    context->connect_return_id = -1;
    context->connect_return_tag = -1;
    context->connect_to = NULL;
    context->spawn_context = NULL;
    context->cred_request[0] = '\0';
    context->password[0] = '\0';
    context->encrypted_password[0] = '\0';
    context->port_str[0] = '\0';
    context->pszChallengeResponse[0] = '\0';
    context->pszCrypt[0] = '\0';
    context->pwd_request[0] = '\0';
    context->session[0] = '\0';
    context->session_header[0] = '\0';
    context->singleton_init_hostname[0] = '\0';
    context->singleton_init_kvsname[0] = '\0';
    context->singleton_init_pm_port = -1;
    context->smpd_pwd[0] = '\0';
#ifdef HAVE_WINDOWS_H
    context->wait.hProcess = NULL;
    context->wait.hThread = NULL;
#else
    context->wait = 0;
#endif
    context->process = NULL;
    context->sspi_header[0] = '\0';
    context->sspi_context = NULL;
    context->sspi_type = SMPD_SSPI_DELEGATE;
    context->sspi_job_key[0] = '\0';
    context->first_output_stderr = SMPD_TRUE;
    context->first_output_stdout = SMPD_TRUE;

    if (sock != MPIDU_SOCK_INVALID_SOCK)
    {
	result = MPIDU_Sock_set_user_ptr(sock, context);
	if (result != MPI_SUCCESS)
	{
	    smpd_err_printf("unable to set the sock user ptr while initializing context,\nsock error: %s\n",
		get_sock_error_string(result));
	}
    }

    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_create_sspi_client_context"
int smpd_create_sspi_client_context(smpd_sspi_client_context_t **new_context)
{
    smpd_sspi_client_context_t *context;

    smpd_enter_fn(FCNAME);

    context = (smpd_sspi_client_context_t *)MPIU_Malloc(sizeof(smpd_sspi_client_context_t));
    if (context == NULL)
    {
	*new_context = NULL;
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    context->buffer = NULL;
    context->buffer_length = 0;
    context->out_buffer = NULL;
    context->out_buffer_length = 0;
    context->max_buffer_size = 0;
#ifdef HAVE_WINDOWS_H
    SecInvalidateHandle(&context->credential);
    SecInvalidateHandle(&context->context);
    memset(&context->expiration_time, 0, sizeof(TimeStamp));
    context->user_handle = INVALID_HANDLE_VALUE;
    context->job = INVALID_HANDLE_VALUE;
    context->flags = 0;
    context->close_handle = SMPD_TRUE;
#endif
    /* FIXME: this insertion needs to be thread safe */
    if (smpd_process.sspi_context_list == NULL)
    {
	context->id = 0;
    }
    else
    {
	context->id = smpd_process.sspi_context_list->id + 1;
    }
    context->next = smpd_process.sspi_context_list;
    smpd_process.sspi_context_list = context;
    *new_context = context;
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_free_sspi_client_context"
int smpd_free_sspi_client_context(smpd_sspi_client_context_t **context)
{
    smpd_sspi_client_context_t *iter, *trailer;

    smpd_enter_fn(FCNAME);

    trailer = iter = smpd_process.sspi_context_list;
    while (iter)
    {
	if (iter == *context)
	{
	    if (trailer != iter)
	    {
		trailer->next = iter->next;
	    }
	    else
	    {
		smpd_process.sspi_context_list = iter->next;
	    }
	    break;
	}
	if (trailer != iter)
	    trailer = trailer->next;
	iter = iter->next;
    }
    if (iter == NULL)
    {
	smpd_dbg_printf("freeing a sspi_client_context not in the global list\n");
    }
    /* FIXME: cleanup sspi structures */
    MPIU_Free(*context);
    *context = NULL;
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_generate_session_header"
int smpd_generate_session_header(char *str, int session_id)
{
    char * str_orig;
    int result;
    int len;

    smpd_enter_fn(FCNAME);

    str_orig = str;
    *str = '\0';
    len = SMPD_MAX_SESSION_HEADER_LENGTH;

    /* add header fields */
    result = MPIU_Str_add_int_arg(&str, &len, "id", session_id);
    if (result != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("unable to create session header, adding session id failed.\n");
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    result = MPIU_Str_add_int_arg(&str, &len, "parent", smpd_process.id);
    if (result != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("unable to create session header, adding parent id failed.\n");
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    result = MPIU_Str_add_int_arg(&str, &len, "level", smpd_process.level + 1);
    if (result != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("unable to create session header, adding session level failed.\n");
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    /* remove the trailing space */
    str--;
    *str = '\0';

    smpd_dbg_printf("session header: (%s)\n", str_orig);
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_get_password"
void smpd_get_password(char *password)
{
#ifdef HAVE_WINDOWS_H
    HANDLE hStdin;
    DWORD dwMode;
#else
    struct termios terminal_settings, original_settings;
#endif
    size_t len;

    smpd_enter_fn(FCNAME);

#ifdef HAVE_WINDOWS_H

    hStdin = GetStdHandle(STD_INPUT_HANDLE);
    if (!GetConsoleMode(hStdin, &dwMode))
	dwMode = ENABLE_LINE_INPUT | ENABLE_PROCESSED_INPUT | ENABLE_MOUSE_INPUT;
    SetConsoleMode(hStdin, dwMode & ~ENABLE_ECHO_INPUT);
    *password = '\0';
    fgets(password, 100, stdin);
    SetConsoleMode(hStdin, dwMode);

    fprintf(stderr, "\n");

#else

    /* save the current terminal settings */
    tcgetattr(STDIN_FILENO, &terminal_settings);
    original_settings = terminal_settings;

    /* turn off echo */
    terminal_settings.c_lflag &= ~ECHO;
    terminal_settings.c_lflag |= ECHONL;
    tcsetattr(STDIN_FILENO, TCSAFLUSH, &terminal_settings);

    /* check that echo is off */
    tcgetattr(STDIN_FILENO, &terminal_settings);
    if (terminal_settings.c_lflag & ECHO)
    {
	smpd_err_printf("\nunable to turn off the terminal echo\n");
	tcsetattr(STDIN_FILENO, TCSANOW, &original_settings);
    }

    fgets(password, 100, stdin);

    /* restore the original settings */
    tcsetattr(STDIN_FILENO, TCSANOW, &original_settings);

#endif

    while ((len = strlen(password)) > 0)
    {
	if (password[len-1] == '\r' || password[len-1] == '\n')
	    password[len-1] = '\0';
	else
	    break;
    }
    smpd_exit_fn(FCNAME);
}

#undef FCNAME
#define FCNAME "smpd_get_account_and_password"
void smpd_get_account_and_password(char *account, char *password)
{
    size_t len;
#ifdef HAVE_WINDOWS_H
    char default_username[100] = "";
    ULONG default_len = 100;
#endif

    smpd_enter_fn(FCNAME);

#ifdef HAVE_WINDOWS_H
    if (!GetUserNameEx(NameSamCompatible, default_username, &default_len))
    {
	default_username[0] = '\0';
    }
#endif

    do
    {
	do
	{
#ifdef HAVE_WINDOWS_H
	    if (default_username[0] != '\0')
	    {
		fprintf(stderr, "account (domain\\user) [%s]: ", default_username);
	    }
	    else
	    {
		fprintf(stderr, "account (domain\\user): ");
	    }
#else
	    fprintf(stderr, "account (domain\\user): ");
#endif
	    fflush(stderr);
	    *account = '\0';
	    fgets(account, 100, stdin);
	    while (strlen(account))
	    {
		len = strlen(account);
		if (account[len-1] == '\r' || account[len-1] == '\n')
		    account[len-1] = '\0';
		else
		    break;
	    }
#ifdef HAVE_WINDOWS_H
	    if ((strlen(account) == 0) && (default_username[0] != '\0'))
	    {
		strcpy(account, default_username);
	    }
#endif
	} 
	while (strlen(account) == 0);

	fprintf(stderr, "password: ");
	fflush(stderr);

	smpd_get_password(password);
	if (strlen(password) == 0)
	{