smpd_handle_command.c

mpi并行计算的c++代码 可用vc或gcc编译通过 可以用来搭建并行计算试验环境

			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)
			{
			    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 = 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)
			{
			    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 = 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);
		    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, "remove_job") == 0)
		{
		    /* print the result of the remove_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, "associate_job") == 0)
		{
		    /* print the result of the associate_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
		{
		    smpd_err_printf("result returned for unhandled '%s' command:\n command: '%s'\n result: '%s'\n", iter->cmd_str, iter->cmd, str);
		}
	    }
	    else
	    {
		smpd_err_printf("no result string in the result command.\n");
	    }
	    if (trailer == iter)
	    {
		context->wait_list = context->wait_list->next;
	    }
	    else
	    {
		trailer->next = iter->next;
	    }
	    result = smpd_free_command(iter);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to free command in the wait_list\n");
	    }
	    smpd_exit_fn(FCNAME);
	    return ret_val;
	}
	else
	{
	    smpd_dbg_printf("tag %d != match_tag %d\n", iter->tag, match_tag);
	}
	if (trailer != iter)
	    trailer = trailer->next;
	iter = iter->next;
    }
    if (context->wait_list == NULL)
    {
	smpd_err_printf("result command received but the wait_list is empty.\n");
    }
    else
    {
	smpd_err_printf("result command did not match any commands in the wait_list.\n");
    }
    smpd_exit_fn(FCNAME);
    return SMPD_FAIL;
}

#undef FCNAME
#define FCNAME "get_name_key_value"