smpd_handle_command.c

fortran并行计算包

                        smpd_err_printf("smpd_free_context failed, error = %d\n", result);
                    }
                    return SMPD_FAIL;
                }
                result = MPIDU_Sock_post_connect(context->set, p_singleton_mpiexec_context,
                                                    connect_to_host,
                                                    smpd_process.singleton_client_port,
                                                    &p_singleton_mpiexec_context->sock);
                if(result != MPI_SUCCESS){
                    context->state = SMPD_DONE;
                    smpd_err_printf("MPIDU_Sock_post_connect failed, error = %s\n", get_sock_error_string(result));
                    result = smpd_free_context(p_singleton_mpiexec_context);
                    if(result != SMPD_SUCCESS){ 
                        smpd_err_printf("smpd_free_context failed, error = %d\n", result);
                    }
                    return SMPD_FAIL;
                }
                ret_val = SMPD_SUCCESS;
            }
            else{
		        printf("%s\n%d\n%s\n", host_description, listener_port, smpd_process.kvs_name);
            }
		    /*printf("%s %d %s\n", smpd_process.host_list->host, smpd_process.port, smpd_process.kvs_name);*/
		    fflush(stdout);
		}
        else if (strcmp(iter->cmd_str, "proc_info") == 0){
            smpd_dbg_printf("%s command result = %s\n", iter->cmd_str, str);
            ret_val = SMPD_DBS_RETURN;
        }
		else if (strcmp(iter->cmd_str, "spawn") == 0)
		{
		    smpd_dbg_printf("%s command result = %s\n", iter->cmd_str, str);
		    ret_val = SMPD_DBS_RETURN;
		}
		else if (iter->cmd_str[0] == 'd' && iter->cmd_str[1] == 'b')
		{
		    smpd_dbg_printf("%s command result = %s\n", iter->cmd_str, str);
		    ret_val = SMPD_DBS_RETURN;
		}
		else if ((strcmp(iter->cmd_str, "init") == 0) || (strcmp(iter->cmd_str, "finalize") == 0))
		{
		    smpd_dbg_printf("%s command result = %s\n", iter->cmd_str, str);
		    ret_val = SMPD_DBS_RETURN;
		}
		else if (strcmp(iter->cmd_str, "barrier") == 0)
		{
		    smpd_dbg_printf("%s command result = %s\n", iter->cmd_str, str);
		    ret_val = SMPD_DBS_RETURN;
		}
		else if (strcmp(iter->cmd_str, "validate") == 0)
		{
		    /* print the result of the validate command */
		    printf("%s\n", str);
		    /* close the session */
		    ret_val = smpd_create_command("done", smpd_process.id, context->id, SMPD_FALSE, &cmd_ptr);
		    if (ret_val == SMPD_SUCCESS)
		    {
			ret_val = smpd_post_write_command(context, cmd_ptr);
			if (ret_val == SMPD_SUCCESS)
			{
			    ret_val = SMPD_CLOSE;
			}
			else
			{
			    smpd_err_printf("unable to post a write of a done command.\n");
			}
		    }
		    else
		    {
			smpd_err_printf("unable to create a done command.\n");
		    }
		}
		else if (strcmp(iter->cmd_str, "status") == 0)
		{
		    /* print the result of the status command */
		    printf("smpd running on %s\n", smpd_process.console_host);
		    if (strcmp(str, "none"))
		    {
			printf("dynamic hosts: %s\n", str);
		    }
		    ret_val = smpd_create_command("done", smpd_process.id, context->id, SMPD_FALSE, &cmd_ptr);
		    if (ret_val == SMPD_SUCCESS)
		    {
			ret_val = smpd_post_write_command(context, cmd_ptr);
			if (ret_val == SMPD_SUCCESS)
			{
			    ret_val = SMPD_CLOSE;
			}
			else
			{
			    smpd_err_printf("unable to post a write of a done command.\n");
			}
		    }
		    else
		    {
			smpd_err_printf("unable to create a done command.\n");
		    }
		}
		else if (strcmp(iter->cmd_str, "get") == 0)
		{
		    if (strcmp(str, SMPD_SUCCESS_STR) == 0)
		    {
			/* print the result of the get command */
			char value[SMPD_MAX_VALUE_LENGTH];
			if (MPIU_Str_get_string_arg(context->read_cmd.cmd, "value", value, SMPD_MAX_VALUE_LENGTH) == MPIU_STR_SUCCESS)
			{
			    printf("%s\n", value);
			}
			else
			{
			    printf("Error: unable to get the value from the result command - '%s'\n", context->read_cmd.cmd);
			}
		    }
		    else
		    {
			printf("%s\n", str);
		    }
		}
		else if (strcmp(iter->cmd_str, "set") == 0 || strcmp(iter->cmd_str, "delete") == 0)
		{
		    /* print the result of the set or delete command */
		    printf("%s\n", str);
		    if (smpd_process.do_console_returns)
		    {
			/* close the session */
			ret_val = smpd_create_command("done", smpd_process.id, context->id, SMPD_FALSE, &cmd_ptr);
			if (ret_val == SMPD_SUCCESS)
			{
			    ret_val = smpd_post_write_command(context, cmd_ptr);
			    if (ret_val == SMPD_SUCCESS)
			    {
				ret_val = SMPD_CLOSE;
			    }
			    else
			    {
				smpd_err_printf("unable to post a write of a done command.\n");
			    }
			}
			else
			{
			    smpd_err_printf("unable to create a done command.\n");
			}
		    }
		}
		else if (strcmp(iter->cmd_str, "stat") == 0)
		{
		    /* print the result of the stat command */
		    printf("%s\n", str);
		}
		else if (strcmp(iter->cmd_str, "cred_request") == 0)
		{
		    if (strcmp(str, SMPD_SUCCESS_STR) == 0)
		    {
			/* FIXME: decrypt the password */
			if (MPIU_Str_get_string_arg(context->read_cmd.cmd, "account", smpd_process.UserAccount, SMPD_MAX_ACCOUNT_LENGTH) == MPIU_STR_SUCCESS &&
			    MPIU_Str_get_string_arg(context->read_cmd.cmd, "password", context->encrypted_password/*smpd_process.UserPassword*/, SMPD_MAX_PASSWORD_LENGTH) == MPIU_STR_SUCCESS)
			{
			    int length = SMPD_MAX_PASSWORD_LENGTH;
			    smpd_dbg_printf("cred_request succeeded, account: '%s'\n", smpd_process.UserAccount);
			    result = smpd_decrypt_data(context->encrypted_password, SMPD_MAX_PASSWORD_LENGTH, smpd_process.UserPassword, &length);
			    if (result == SMPD_SUCCESS)
			    {
				strcpy(iter->context->account, smpd_process.UserAccount);
				strcpy(iter->context->password, smpd_process.UserPassword);
				strcpy(iter->context->cred_request, "yes");
				iter->context->read_state = SMPD_IDLE;
				iter->context->write_state = SMPD_WRITING_CRED_ACK_YES;
				result = MPIDU_Sock_post_write(iter->context->sock, iter->context->cred_request, SMPD_MAX_CRED_REQUEST_LENGTH, SMPD_MAX_CRED_REQUEST_LENGTH, NULL);
			    }
			    ret_val = result == MPI_SUCCESS ? SMPD_SUCCESS : SMPD_FAIL;
			}
			else
			{
			    smpd_err_printf("invalid cred_request result returned, no account and password specified: '%s'\n", context->read_cmd.cmd);
			    ret_val = SMPD_FAIL;
			}
		    }
		    else if (strcmp(str, "sspi") == 0)
		    {
			strcpy(iter->context->cred_request, SMPD_CRED_ACK_SSPI);
			iter->context->read_state = SMPD_IDLE;
			iter->context->write_state = SMPD_WRITING_CRED_ACK_SSPI;
			result = MPIDU_Sock_post_write(iter->context->sock, iter->context->cred_request, SMPD_MAX_CRED_REQUEST_LENGTH, SMPD_MAX_CRED_REQUEST_LENGTH, NULL);
			ret_val = result == MPI_SUCCESS ? SMPD_SUCCESS : SMPD_FAIL;
		    }
		    else if (strcmp(str, "sspi_job") == 0)
		    {
			strcpy(iter->context->cred_request, SMPD_CRED_ACK_SSPI_JOB_KEY);
			iter->context->read_state = SMPD_IDLE;
			iter->context->write_state = SMPD_WRITING_CRED_ACK_SSPI_JOB_KEY;
			result = MPIDU_Sock_post_write(iter->context->sock, iter->context->cred_request, SMPD_MAX_CRED_REQUEST_LENGTH, SMPD_MAX_CRED_REQUEST_LENGTH, NULL);
			ret_val = result == MPI_SUCCESS ? SMPD_SUCCESS : SMPD_FAIL;
		    }
		    else
		    {
			strcpy(iter->context->cred_request, "no");
			iter->context->read_state = SMPD_IDLE;
			iter->context->write_state = SMPD_WRITING_CRED_ACK_NO;
			result = MPIDU_Sock_post_write(iter->context->sock, iter->context->cred_request, SMPD_MAX_CRED_REQUEST_LENGTH, SMPD_MAX_CRED_REQUEST_LENGTH, NULL);
			ret_val = result == MPI_SUCCESS ? SMPD_SUCCESS : SMPD_FAIL;
		    }
		}
		else if (strcmp(iter->cmd_str, "sspi_init") == 0)
		{
		    if (strcmp(str, SMPD_SUCCESS_STR) == 0)
		    {
			/* decode the sspi buffer and context pointer from the command */
			MPIU_Str_get_string_arg(context->read_cmd.cmd, "sspi_context", context_str, 20);
			if (sscanf(context_str, "%p", &orig_context) != 1)
			{
			    smpd_err_printf("unable to decode a pointer from this string: '%s'\n", context_str);
			    smpd_exit_fn(FCNAME);
			    return SMPD_FAIL;
			}
			MPIU_Str_get_int_arg(context->read_cmd.cmd, "sspi_id", &orig_context->sspi_context->id);
			MPIU_Str_get_int_arg(context->read_cmd.cmd, "data_length", &orig_context->sspi_context->buffer_length);
			if (orig_context->sspi_context->buffer != NULL)
			{
			    MPIU_Free(orig_context->sspi_context->buffer);
			    orig_context->sspi_context->buffer = NULL;
			}
			if (orig_context->sspi_context->buffer_length > 0)
			{
			    orig_context->sspi_context->buffer = MPIU_Malloc(orig_context->sspi_context->buffer_length);
			    if (orig_context->sspi_context->buffer == NULL)
			    {
				smpd_err_printf("unable to allocate %d bytes for an sspi buffer\n", orig_context->sspi_context->buffer_length);
				smpd_exit_fn(FCNAME);
				return SMPD_FAIL;
			    }
			}
			MPIU_Str_get_string_arg(context->read_cmd.cmd, "data", str, SMPD_MAX_CMD_LENGTH);
			smpd_decode_buffer(str, orig_context->sspi_context->buffer, orig_context->sspi_context->buffer_length, &num_decoded);
			/* send the sspi buffer to the decoded context */
			/* move the state of that context to WRITING_SSPI_HEADER */
			orig_context->read_state = SMPD_IDLE;
			orig_context->write_state = SMPD_WRITING_CLIENT_SSPI_HEADER;
			MPIU_Snprintf(orig_context->sspi_header, SMPD_SSPI_HEADER_LENGTH, "%d", num_decoded);
			result = MPIDU_Sock_post_write(orig_context->sock, orig_context->sspi_header, SMPD_SSPI_HEADER_LENGTH, SMPD_SSPI_HEADER_LENGTH, NULL);
			if (result == MPI_SUCCESS)
			{
			    ret_val = SMPD_SUCCESS;
			}
			else
			{
#ifdef HAVE_WINDOWS_H
			    smpd_process.sec_fn->DeleteSecurityContext(&orig_context->sspi_context->context);
			    smpd_process.sec_fn->FreeCredentialsHandle(&orig_context->sspi_context->credential);
#endif
			    smpd_err_printf("unable to post a write of the sspi header,\nsock error: %s\n", get_sock_error_string(result));
			    orig_context->state = SMPD_CLOSING;
			    result = MPIDU_Sock_post_close(orig_context->sock);
			    smpd_exit_fn(FCNAME);
			    ret_val = (result == MPI_SUCCESS) ? SMPD_SUCCESS : SMPD_FAIL;
			}
		    }
		    else
		    {
			/* close the sspi context */
			/* cleanup the sspi structures */
			/* notify connection failure */
		    }
		}
		else if (strcmp(iter->cmd_str, "sspi_iter") == 0)
		{
		    if (strcmp(str, SMPD_SUCCESS_STR) == 0)
		    {
			/* decode the sspi buffer and context pointer from the command */
			MPIU_Str_get_string_arg(context->read_cmd.cmd, "sspi_context", context_str, 20);
			if (sscanf(context_str, "%p", &orig_context) != 1)
			{
			    smpd_err_printf("unable to decode a pointer from this string: '%s'\n", context_str);
			    smpd_exit_fn(FCNAME);
			    return SMPD_FAIL;
			}
			MPIU_Str_get_int_arg(context->read_cmd.cmd, "data_length", &orig_context->sspi_context->buffer_length);
			if (orig_context->sspi_context->buffer != NULL)
			{
			    MPIU_Free(orig_context->sspi_context->buffer);
			    orig_context->sspi_context->buffer = NULL;
			}
			if (orig_context->sspi_context->buffer_length > 0)
			{
			    orig_context->sspi_context->buffer = MPIU_Malloc(orig_context->sspi_context->buffer_length);
			    if (orig_context->sspi_context->buffer == NULL)
			    {
				smpd_err_printf("unable to allocate %d bytes for an sspi buffer\n", orig_context->sspi_context->buffer_length);
				smpd_exit_fn(FCNAME);
				return SMPD_FAIL;
			    }
			    MPIU_Str_get_string_arg(context->read_cmd.cmd, "data", str, SMPD_MAX_CMD_LENGTH);
			    smpd_decode_buffer(str, orig_context->sspi_context->buffer, orig_context->sspi_context->buffer_length, &num_decoded);
			    /* send the sspi buffer to the decoded context */
			    /* move the state of that context to WRITING_SSPI_HEADER */
			    orig_context->read_state = SMPD_IDLE;
			    orig_context->write_state = SMPD_WRITING_CLIENT_SSPI_HEADER;
			    MPIU_Snprintf(orig_context->sspi_header, SMPD_SSPI_HEADER_LENGTH, "%d", num_decoded);
			    result = MPIDU_Sock_post_write(orig_context->sock, orig_context->sspi_header, SMPD_SSPI_HEADER_LENGTH, SMPD_SSPI_HEADER_LENGTH, NULL);
			    if (result == MPI_SUCCESS)
			    {
				ret_val = SMPD_SUCCESS;
			    }
			    else
			    {
#ifdef HAVE_WINDOWS_H
				smpd_process.sec_fn->DeleteSecurityContext(&orig_context->sspi_context->context);
				smpd_process.sec_fn->FreeCredentialsHandle(&orig_context->sspi_context->credential);
#endif
				smpd_err_printf("unable to post a write of the sspi header,\nsock error: %s\n", get_sock_error_string(result));
				orig_context->state = SMPD_CLOSING;
				result = MPIDU_Sock_post_close(orig_context->sock);
				smpd_exit_fn(FCNAME);
				ret_val = (result == MPI_SUCCESS) ? SMPD_SUCCESS : SMPD_FAIL;
			    }
			}
			else
			{
			    if (orig_context->sspi_context->buffer_length == 0)
			    {
				/* no more data, empty buffer returned */
				/* FIXME: This assumes that the server knows to post a write of the delegate command because it knows that no buffer will be returned by the iter command */
				orig_context->write_state = SMPD_IDLE;
				orig_context->read_state = SMPD_READING_CLIENT_SSPI_HEADER;
				result = MPIDU_Sock_post_read(orig_context->sock, orig_context->sspi_header, SMPD_SSPI_HEADER_LENGTH, SMPD_SSPI_HEADER_LENGTH, NULL);
				if (result != MPI_SUCCESS)
				{
				    /* FIXME: Add code to cleanup sspi structures */
				    smpd_err_printf("unable to post a read of the client sspi header,\nsock error: %s\n", get_sock_error_string(result));
				    orig_context->state = SMPD_CLOSING;
				    result = MPIDU_Sock_post_close(orig_context->sock);
				    smpd_exit_fn(FCNAME);
				    return (result == MPI_SUCCESS) ? SMPD_SUCCESS : SMPD_FAIL;
				}
			    }
			    else
			    {
				/* error decoding the data_length parameter */
#ifdef HAVE_WINDOWS_H
				smpd_process.sec_fn->DeleteSecurityContext(&orig_context->sspi_context->context);
				smpd_process.sec_fn->FreeCredentialsHandle(&orig_context->sspi_context->credential);
#endif
				smpd_err_printf("unable to decode the data_length parameter,\n");
				orig_context->state = SMPD_CLOSING;
				result = MPIDU_Sock_post_close(orig_context->sock);
				smpd_exit_fn(FCNAME);
				ret_val = (result == MPI_SUCCESS) ? SMPD_SUCCESS : SMPD_FAIL;
			    }
			}
		    }
		    else
		    {
			/* close the sspi context */
			/* cleanup the sspi structures */
			/* notify connection failure */
		    }
		}
		else if (strcmp(iter->cmd_str, "exit_on_done") == 0)
		{
		    ret_val = SMPD_DBS_RETURN;
		    /*
		    if (strcmp(str, SMPD_SUCCESS_STR) == 0)
		    {
			ret_val = SMPD_SUCCESS;
		    }
		    else
		    {
			smpd_err_printf("exit_on_done failed: %s\n", str);
			ret_val = SMPD_ABORT;
		    }
		    */
		}
		else if (strcmp(iter->cmd_str, "suspend") == 0)
		{
		    smpd_dbg_printf("suspend command result returned: %s\n", str);
		    ret_val = smpd_handle_suspend_result(iter, str);
		}
		else if (strcmp(iter->cmd_str, "add_job") == 0)
		{
		    /* print the result of the add_job command */
		    printf("%s\n", str);
		    /* close the session */
		    ret_val = smpd_create_command("done", smpd_process.id, context->id, SMPD_FALSE, &cmd_ptr);
		    if (ret_val == SMPD_SUCCESS)
		    {
			ret_val = smpd_post_write_command(context, cmd_ptr);
			if (ret_val == SMPD_SUCCESS)
			{
			    ret_val = SMPD_CLOSE;
			}
			else
			{
			    smpd_err_printf("unable to post a write of a done command.\n");
			}
		    }
		    else
		    {
			smpd_err_printf("unable to create a done command.\n");
		    }
		}
		else if (strcmp(iter->cmd_str, "add_job_and_password") == 0)
		{
		    /* print the result of the add_job_and_password command */
		    printf("%s\n", str);
		    /* close the session */
		    ret_val = smpd_create_command("done", smpd_process.id, context->id, SMPD_FALSE, &cmd_ptr);