smpd_command.c

fortran并行计算包

    if (result != SMPD_SUCCESS)
    {
	*context_pptr = NULL;
	MPIU_Free(context);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    
    /* add the context to the global list */
    context->next = smpd_process.context_list;
    smpd_process.context_list = context;

    *context_pptr = context;
    smpd_exit_fn(FCNAME);
    return result;
}

/*#define DEBUG_SMPD_FREE_CONTEXT*/

#ifdef DEBUG_SMPD_FREE_CONTEXT
typedef struct cfree_t
{
    smpd_context_t *context;
    struct cfree_t *next;
} cfree_t;
static cfree_t *free_list = NULL;
#endif

#undef FCNAME
#define FCNAME "smpd_free_context"
int smpd_free_context(smpd_context_t *context)
{
    SMPD_BOOL found = SMPD_FALSE;
    smpd_context_t *iter, *trailer;

    smpd_enter_fn(FCNAME);
    if (context)
    {
#ifdef DEBUG_SMPD_FREE_CONTEXT
	/* check debugging free list */
	cfree_t *citer = free_list;
	while (citer)
	{
	    if (citer->context == context)
	    {
		smpd_err_printf("%s context freed twice.\n", smpd_get_context_str(context));
	    }
	    citer = citer->next;
	}
#endif

	/* remove the context from the global list */
	iter = trailer = smpd_process.context_list;
	while (iter)
	{
	    if (iter == context)
	    {
		if (iter == smpd_process.context_list)
		    smpd_process.context_list = smpd_process.context_list->next;
		else
		    trailer->next = iter->next;
		found = SMPD_TRUE;
		break;
	    }
	    if (trailer != iter)
		trailer = trailer->next;
	    iter = iter->next;
	}

	if (!found)
	{
#ifdef DEBUG_SMPD_FREE_CONTEXT
	    smpd_dbg_printf("freeing a %s context not in the global list - this should be impossible.\n",
		smpd_get_context_str(context));
#else
	    smpd_dbg_printf("freeing a context not in the global list - this should be impossible.\n");
	    smpd_exit_fn(FCNAME);
	    return SMPD_SUCCESS;
#endif
	}

        smpd_dbg_printf("freeing %s context.\n", smpd_get_context_str(context));

	/* this check isn't full-proof because random data might match SMPD_CONTEXT_FREED */
	/* This also doesn't work because free clobbers the memory instead of leaving it alone */
	if (context->type == SMPD_CONTEXT_FREED)
	{
	    smpd_err_printf("attempt to free context more than once.\n");
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}

	/* remove any references to this context in the process structure */
	if (context->process)
	{
	    switch (context->type)
	    {
	    case SMPD_CONTEXT_STDIN:
		context->process->in = NULL;
		break;
	    case SMPD_CONTEXT_STDOUT:
	    case SMPD_CONTEXT_STDOUT_RSH:
		context->process->out = NULL;
		break;
	    case SMPD_CONTEXT_STDERR:
	    case SMPD_CONTEXT_STDERR_RSH:
		context->process->err = NULL;
		break;
	    case SMPD_CONTEXT_PMI:
		context->process->pmi = NULL;
		break;
	    }
	}

	/* erase the contents to help track down use of freed structures */
	/* This doesn't work because free clobbers the memory instead of leaving it alone */
	memset(context, 0, sizeof(smpd_context_t));
	smpd_init_context(context, SMPD_CONTEXT_FREED, MPIDU_SOCK_INVALID_SET, MPIDU_SOCK_INVALID_SOCK, -1);

#ifdef DEBUG_SMPD_FREE_CONTEXT
	/* add to debugging free list */
	citer = (cfree_t*)malloc(sizeof(cfree_t));
	citer->context = context;
	citer->next = free_list;
	free_list = citer;
#else
	MPIU_Free(context);
#endif
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_add_command_arg"
int smpd_add_command_arg(smpd_command_t *cmd_ptr, char *param, char *value)
{
    char *str;
    int len;
    int result;
    int cmd_length;

    smpd_enter_fn(FCNAME);

    cmd_length = (int)strlen(cmd_ptr->cmd);
    if (cmd_length > SMPD_MAX_CMD_LENGTH)
    {
	smpd_err_printf("invalid cmd string length: %d\n", cmd_length);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    len = (int)(SMPD_MAX_CMD_LENGTH - cmd_length);
    str = &cmd_ptr->cmd[cmd_length];

    /* make sure there is a space after the last parameter in the command */
    if (cmd_length > 0)
    {
	if (cmd_ptr->cmd[cmd_length-1] != MPIU_STR_SEPAR_CHAR)
	{
	    if (len < 2)
	    {
		smpd_err_printf("unable to add the command parameter: %s=%s\n", param, value);
		smpd_exit_fn(FCNAME);
		return SMPD_FAIL;
	    }
	    cmd_ptr->cmd[cmd_length] = MPIU_STR_SEPAR_CHAR;
	    len--;
	    str++;
	}
    }

    result = MPIU_Str_add_string_arg(&str, &len, param, value);
    if (result != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("unable to add the command parameter: %s=%s\n", param, value);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_add_command_int_arg"
int smpd_add_command_int_arg(smpd_command_t *cmd_ptr, char *param, int value)
{
    char *str;
    int len;
    int result;
    int cmd_length;

    smpd_enter_fn(FCNAME);

    cmd_length = (int)strlen(cmd_ptr->cmd);
    if (cmd_length > SMPD_MAX_CMD_LENGTH)
    {
	smpd_err_printf("invalid cmd string length: %d\n", cmd_length);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    len = (int)(SMPD_MAX_CMD_LENGTH - cmd_length);
    str = &cmd_ptr->cmd[cmd_length];

    /* make sure there is a space after the last parameter in the command */
    if (cmd_length > 0)
    {
	if (cmd_ptr->cmd[cmd_length-1] != MPIU_STR_SEPAR_CHAR)
	{
	    if (len < 2)
	    {
		smpd_err_printf("unable to add the command parameter: %s=%d\n", param, value);
		smpd_exit_fn(FCNAME);
		return SMPD_FAIL;
	    }
	    cmd_ptr->cmd[cmd_length] = MPIU_STR_SEPAR_CHAR;
	    len--;
	    str++;
	}
    }

    result = MPIU_Str_add_int_arg(&str, &len, param, value);
    if (result != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("unable to add the command parameter: %s=%d\n", param, value);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_add_command_binary_arg"
int smpd_add_command_binary_arg(smpd_command_t *cmd_ptr, char *param, void *buffer, int length)
{
    char *str;
    int len;
    int result;
    int cmd_length;
    int saved_length;

    smpd_enter_fn(FCNAME);

    cmd_length = (int)strlen(cmd_ptr->cmd);
    if (cmd_length > SMPD_MAX_CMD_LENGTH)
    {
	smpd_err_printf("invalid cmd string length: %d\n", cmd_length);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    len = (int)(SMPD_MAX_CMD_LENGTH - cmd_length);
    str = &cmd_ptr->cmd[cmd_length];

    /* make sure there is a space after the last parameter in the command */
    if (cmd_length > 0)
    {
	if (cmd_ptr->cmd[cmd_length-1] != MPIU_STR_SEPAR_CHAR)
	{
	    if (len < 2)
	    {
		smpd_err_printf("unable to add the command parameter: %s=%d byte buffer\n", param, length);
		smpd_exit_fn(FCNAME);
		return SMPD_FAIL;
	    }
	    cmd_ptr->cmd[cmd_length] = MPIU_STR_SEPAR_CHAR;
	    len--;
	    str++;
	}
    }

    saved_length = len;
    result = MPIU_Str_add_binary_arg(&str, &len, param, buffer, length);
    if (result != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("unable to add the command parameter: %s=%d byte buffer won't fit in %d character length string\n", param, length, saved_length);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_forward_command"
int smpd_forward_command(smpd_context_t *src, smpd_context_t *dest)
{
    int result;
    smpd_command_t *cmd;

    smpd_enter_fn(FCNAME);

    result = smpd_create_command_copy(&src->read_cmd, &cmd);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("unable to create a copy of the command to forward.\n");
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    smpd_dbg_printf("posting write of forwarded command: \"%s\"\n", cmd->cmd);
    result = smpd_post_write_command(dest, cmd);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("unable to post a write of a forwarded command.\n");
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_post_read_command"
int smpd_post_read_command(smpd_context_t *context)
{
    int result;

    smpd_enter_fn(FCNAME);

    /* post a read for the next command header */
    smpd_dbg_printf("posting a read for a command header on the %s context, sock %d\n", smpd_get_context_str(context), MPIDU_Sock_get_sock_id(context->sock));
    context->read_state = SMPD_READING_CMD_HEADER;
    context->read_cmd.state = SMPD_CMD_READING_HDR;
    result = MPIDU_Sock_post_read(context->sock, context->read_cmd.cmd_hdr_str, SMPD_CMD_HDR_LENGTH, SMPD_CMD_HDR_LENGTH, NULL);
    if (result != MPI_SUCCESS)
    {
	smpd_err_printf("unable to post a read for the next command header,\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_post_write_command"
int smpd_post_write_command(smpd_context_t *context, smpd_command_t *cmd)
{
    int result;
    smpd_command_t *iter;

    smpd_enter_fn(FCNAME);

    if (context == NULL)
    {
	smpd_dbg_printf("unable to post a write of command '%s' on a NULL context", cmd->cmd);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    smpd_package_command(cmd);
    /*smpd_dbg_printf("command after packaging: \"%s\"\n", cmd->cmd);*/
    cmd->next = NULL;
    cmd->state = SMPD_CMD_WRITING_CMD;
    context->write_state = SMPD_WRITING_CMD;

    if (!context->write_list)
    {
	context->write_list = cmd;
    }
    else
    {
	smpd_dbg_printf("enqueueing write at the end of the list.\n");
	iter = context->write_list;
	while (iter->next)
	    iter = iter->next;
	iter->next = cmd;
	smpd_exit_fn(FCNAME);
	return SMPD_SUCCESS;
    }

    cmd->iov[0].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)cmd->cmd_hdr_str;
    cmd->iov[0].MPID_IOV_LEN = SMPD_CMD_HDR_LENGTH;
    cmd->iov[1].MPID_IOV_BUF = (MPID_IOV_BUF_CAST)cmd->cmd;
    cmd->iov[1].MPID_IOV_LEN = cmd->length;
    /*smpd_dbg_printf("command at this moment: \"%s\"\n", cmd->cmd);*/
    smpd_dbg_printf("smpd_post_write_command on the %s context sock %d: %d bytes for command: \"%s\"\n",
	smpd_get_context_str(context), MPIDU_Sock_get_sock_id(context->sock),
	cmd->iov[0].MPID_IOV_LEN + cmd->iov[1].MPID_IOV_LEN,
	cmd->cmd);
    result = MPIDU_Sock_post_writev(context->sock, cmd->iov, 2, NULL);
    if (result != MPI_SUCCESS)
    {
	smpd_err_printf("unable to post a write for the next command,\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_package_command"
int smpd_package_command(smpd_command_t *cmd)
{
    int length;

    smpd_enter_fn(FCNAME);

    /* create the command header - for now it is simply the length of the command string */
    length = (int)strlen(cmd->cmd) + 1;
    if (length > SMPD_MAX_CMD_LENGTH)
    {
	smpd_err_printf("unable to package invalid command of length %d\n", length);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    snprintf(cmd->cmd_hdr_str, SMPD_CMD_HDR_LENGTH, "%d", length);
    cmd->length = length;

    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}