root/branches/Orange-Elaine-Distr-Dir-Branch/src/client/sysint/sys-remove.sm @ 8418

/* 
 * (C) 2003 Clemson University and The University of Chicago 
 *
 * See COPYING in top-level directory.
 */

/** \file
 *  \ingroup sysint
 *
 *  PVFS2 system interface routines for removing an object and its
 *  associated directory entry.
 */

#include <string.h>
#include <assert.h>

#include "client-state-machine.h"
#include "pint-util.h"
#include "pvfs2-debug.h"
#include "job.h"
#include "gossip.h"
#include "str-utils.h"
#include "pint-cached-config.h"
#include "PINT-reqproto-encode.h"
#include "ncache.h"
#include "pvfs2-internal.h"

/*
  PVFS_{i}sys_remove takes the following steps:

  - rmdirent the entry from the parent directory
  - getattr on the object to be removed
  - if the object is a directory
    - check if the dir is empty
    - if so, continue
    - if not, crdirent the previously rmdirent'd
      entry and return -PVFS_ENOTEMPTY
  - if the object is a metafile
    - remove all associated data files
  - remove the actual object specified
*/

extern job_context_id pint_client_sm_context;

enum
{
    RMDIRENT_RETRY = 1,
    CRDIRENT_RETRY,
    RETURN_STORED_ERROR_CODE
};

static int remove_rmdirent_comp_fn(
    void *v_p, struct PVFS_server_resp *resp_p, int i);
static int remove_crdirent_comp_fn(
    void *v_p, struct PVFS_server_resp *resp_p, int i);

#define PRINT_REMOVE_WARNING()                                        \
do {                                                                  \
    gossip_err("WARNING: PVFS_sys_remove() encountered an error "     \
               "which may lead to\n  inconsistent state.\n");         \
    gossip_err("WARNING: PVFS2 fsck (if available) may be needed.\n");\
} while(0)

%%

machine pvfs2_client_remove_sm
{
    state init
    {
        run remove_init;
        default => init_getattr;
    }

    state init_getattr
    {
        jump pvfs2_client_getattr_sm;
        success => rmdirent_setup_msgpair;
        default => cleanup;
    }

    state rmdirent_setup_msgpair
    {
        run remove_rmdirent_setup_msgpair;
        success => rmdirent_xfer_msgpair;
        default => rmdirent_retry_or_fail;
    }

    state rmdirent_xfer_msgpair
    {
        jump pvfs2_msgpairarray_sm;
        success => do_remove;
        default => rmdirent_retry_or_fail;
    }

    state rmdirent_retry_or_fail
    {
        run remove_rmdirent_retry_or_fail;
        RMDIRENT_RETRY => rmdirent_timer;
        default => cleanup;
    }

    state do_remove
    {
        jump pvfs2_client_remove_helper_sm;
        default => check_error_code;
    }

    state check_error_code
    {
        run remove_check_error_code;
        success => cleanup;
        RETURN_STORED_ERROR_CODE => cleanup;
        default => getattr_setup;
    }

    state crdirent_timer
    {
        run remove_generic_timer;
        default => getattr_setup;
    }

    state rmdirent_timer
    {
        run remove_generic_timer;
        default => rmdirent_setup_msgpair;
    }

    state getattr_setup
    {
        run remove_getattr_setup;
        default => parent_getattr;
    }

    state parent_getattr
    {
        jump pvfs2_client_getattr_sm;
        success => crdirent_setup_msgpair;
        default => cleanup;
    }

    state crdirent_setup_msgpair
    {
        run remove_crdirent_setup_msgpair;
        success => crdirent_xfer_msgpair;
        default => crdirent_retry_or_fail;
    }

    state crdirent_xfer_msgpair
    {
        jump pvfs2_msgpairarray_sm;
        success => cleanup;
        default => crdirent_retry_or_fail;
    }

    state crdirent_retry_or_fail
    {
        run remove_crdirent_retry_or_fail;
        CRDIRENT_RETRY => crdirent_timer;
        default => cleanup;
    }

    state cleanup
    {
        run remove_cleanup;
        default => terminate;
    }
}

%%

/** Initiate removal of an object and its directory entry.
 */
PVFS_error PVFS_isys_remove(
    char *object_name,
    PVFS_object_ref parent_ref, 
    const PVFS_credentials *credentials,
    PVFS_sys_op_id *op_id,
    PVFS_hint hints,
    void *user_ptr)
{
    PVFS_error ret = -PVFS_EINVAL;
    PINT_smcb *smcb = NULL;
    PINT_client_sm *sm_p = NULL;

    gossip_debug(GOSSIP_CLIENT_DEBUG, "PVFS_isys_remove entered\n");

    if ((parent_ref.handle == PVFS_HANDLE_NULL) ||
        (parent_ref.fs_id == PVFS_FS_ID_NULL) ||
        (object_name == NULL))
    {
        gossip_err("invalid (NULL) required argument\n");
        return ret;
    }

    PINT_smcb_alloc(&smcb, PVFS_SYS_REMOVE,
             sizeof(struct PINT_client_sm),
             client_op_state_get_machine,
             client_state_machine_terminate,
             pint_client_sm_context);
    if (smcb == NULL)
    {
        return -PVFS_ENOMEM;
    }
    sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);

    PINT_init_msgarray_params(sm_p, parent_ref.fs_id);
    PINT_init_sysint_credentials(sm_p->cred_p, credentials);
    sm_p->u.remove.object_name = object_name;
    sm_p->parent_ref  = parent_ref;
    sm_p->u.remove.stored_error_code = 0;
    PVFS_hint_copy(hints, &sm_p->hints);
    PVFS_hint_add(&sm_p->hints, PVFS_HINT_HANDLE_NAME, sizeof(PVFS_handle), &parent_ref.handle);

    gossip_debug(
        GOSSIP_CLIENT_DEBUG, "Trying to remove entry %s under %llu,%d\n",
        object_name, llu(parent_ref.handle), parent_ref.fs_id);

    /* NOTE: This state machine previously multiplied the default job timeout
     * by five to allow for potentially long sync delays.  We instead now set
     * the default client BMI timeout higher for all operations: if a sync can 
     * go slow then so can any other arbitrary operation queued behind it. -PHC
     */
    
    return PINT_client_state_machine_post(
        smcb,  op_id, user_ptr);
}

/** Remove an object and its directory entry.
 */
PVFS_error PVFS_sys_remove(
    char *object_name,
    PVFS_object_ref parent_ref, 
    const PVFS_credentials *credentials,
    PVFS_hint hints)
{
    PVFS_error ret = -PVFS_EINVAL, error = 0;
    PVFS_sys_op_id op_id;

    gossip_debug(GOSSIP_CLIENT_DEBUG, "PVFS_sys_remove entered\n");

    ret = PVFS_isys_remove(object_name, parent_ref,
                           credentials, &op_id, hints, NULL);
    if (ret)
    {
        PVFS_perror_gossip("PVFS_isys_remove call", ret);
        error = ret;
    }
    else
    {
        ret = PVFS_sys_wait(op_id, "remove", &error);
        if (ret)
        {
            PVFS_perror_gossip("PVFS_sys_wait call", ret);
            error = ret;
        }
    }

    PINT_sys_release(op_id);
    return error;
}


/****************************************************************/

static PINT_sm_action remove_init(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);

    gossip_debug(GOSSIP_CLIENT_DEBUG, "remove state: init\n");

    assert(js_p->error_code == 0);

    PINT_SM_GETATTR_STATE_FILL(
        sm_p->getattr,
        sm_p->parent_ref,
        PVFS_ATTR_COMMON_ALL|PVFS_ATTR_DIR_DIRENT_FILES,
        PVFS_TYPE_DIRECTORY,
        0);

    return SM_ACTION_COMPLETE;
}

static PINT_sm_action remove_rmdirent_setup_msgpair(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -PVFS_EINVAL;
    PINT_sm_msgpair_state *msg_p = NULL;

    /* NOTE: we remove the dirent first because this gets the object
     * out of the system fastest.
     */
    gossip_debug(GOSSIP_CLIENT_DEBUG,
                 "remove state: rmdirent_setup_msgpair\n");

    js_p->error_code = 0;

    PINT_msgpair_init(&sm_p->msgarray_op);
    msg_p = &sm_p->msgarray_op.msgpair;

    /* TODO: Need to find the correct dirent_handle */
    PINT_SERVREQ_RMDIRENT_FILL(
        msg_p->req,
        *sm_p->cred_p,
        sm_p->parent_ref.fs_id,
        sm_p->parent_ref.handle,
        sm_p->getattr.attr.u.dir.dirent_handle[0],
        sm_p->u.remove.object_name,
        sm_p->hints);

    gossip_debug(GOSSIP_REMOVE_DEBUG, "- doing RMDIRENT on %s "
                 "under %llu,%d\n", sm_p->u.remove.object_name,
                 llu(sm_p->parent_ref.handle),
                 sm_p->parent_ref.fs_id);

    msg_p->fs_id = sm_p->parent_ref.fs_id;
    msg_p->handle = sm_p->parent_ref.handle;
    msg_p->retry_flag = PVFS_MSGPAIR_NO_RETRY;
    msg_p->comp_fn = remove_rmdirent_comp_fn;

    ret = PINT_cached_config_map_to_server(
        &msg_p->svr_addr, msg_p->handle, msg_p->fs_id);

    if (ret)
    {
        gossip_err("Failed to map meta server address\n");
        js_p->error_code = ret;
    }

    PINT_sm_push_frame(smcb, 0, &sm_p->msgarray_op);
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action remove_getattr_setup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);

    gossip_debug(GOSSIP_CLIENT_DEBUG,
                 "remove state: getattr_setup\n");

    sm_p->u.remove.stored_error_code = js_p->error_code;
    js_p->error_code = 0;

    PINT_SM_GETATTR_STATE_FILL(
        sm_p->getattr,
        sm_p->object_ref,
        PVFS_ATTR_DIR_DIRENT_FILES,
        PVFS_TYPE_DIRECTORY,
        0);

    return SM_ACTION_COMPLETE;
}

static PINT_sm_action remove_crdirent_setup_msgpair(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_object_attr *attr = NULL;
    int ret = -PVFS_EINVAL;
    PINT_sm_msgpair_state *msg_p = NULL;

    gossip_debug(GOSSIP_CLIENT_DEBUG,
                 "remove state: crdirent_setup_msgpair\n");

    sm_p->u.remove.stored_error_code = js_p->error_code;

    js_p->error_code = 0;

    attr = &sm_p->getattr.attr;
    assert(attr);

    /* TODO: Need to find the correct dirent_handle */
    gossip_debug(GOSSIP_CLIENT_DEBUG, "dirent_handle: %llu\n",
        llu(attr->u.dir.dirent_handle[0]));
 
    PINT_msgpair_init(&sm_p->msgarray_op);
    msg_p = &sm_p->msgarray_op.msgpair;

    PINT_SERVREQ_CRDIRENT_FILL(
        msg_p->req,
        *sm_p->cred_p,
        sm_p->u.remove.object_name,
        sm_p->object_ref.handle,
        sm_p->parent_ref.handle,
        attr->u.dir.dirent_handle[0],
        sm_p->parent_ref.fs_id,
        sm_p->hints);

    gossip_debug(GOSSIP_REMOVE_DEBUG, "- doing CRDIRENT of %s (%llu,%d) "
                 "under %llu,%d\n", sm_p->u.remove.object_name,
                 llu(sm_p->object_ref.handle),
                 sm_p->object_ref.fs_id,
                 llu(sm_p->parent_ref.handle),
                 sm_p->parent_ref.fs_id);

    msg_p->fs_id = sm_p->parent_ref.fs_id;
    msg_p->handle = sm_p->parent_ref.handle;
    msg_p->retry_flag = PVFS_MSGPAIR_NO_RETRY;
    msg_p->comp_fn = remove_crdirent_comp_fn;

    ret = PINT_cached_config_map_to_server(
        &msg_p->svr_addr, msg_p->handle, msg_p->fs_id);

    if (ret)
    {
        gossip_err("Failed to map meta server address\n");
        js_p->error_code = ret;
    }

    PINT_sm_push_frame(smcb, 0, &sm_p->msgarray_op);
    return SM_ACTION_COMPLETE;
}

static int remove_rmdirent_comp_fn(
    void *v_p,
    struct PVFS_server_resp *resp_p,
    int index)
{
    PINT_smcb *smcb = v_p;
    PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_MSGPAIR_PARENT_SM);

    assert(resp_p->op == PVFS_SERV_RMDIRENT);

    if (resp_p->status == 0)
    {
        assert(resp_p->u.rmdirent.entry_handle != PVFS_HANDLE_NULL);
        assert(sm_p->parent_ref.fs_id != PVFS_FS_ID_NULL);

        /* pull handle out of response, also copy in fs_id from before */
        sm_p->object_ref.handle = resp_p->u.rmdirent.entry_handle;
        sm_p->object_ref.fs_id  = sm_p->parent_ref.fs_id;

        gossip_debug(
            GOSSIP_CLIENT_DEBUG,
            "  remove_rmdirent_comp_fn: metafile handle = %llu\n",
            llu(sm_p->object_ref.handle));
    }
    return resp_p->status;
}

static int remove_crdirent_comp_fn(
    void *v_p,
    struct PVFS_server_resp *resp_p,
    int index)
{
    assert(resp_p->op == PVFS_SERV_CRDIRENT);

    if (resp_p->status == 0)
    {
        gossip_debug(GOSSIP_CLIENT_DEBUG,
                     "  remove_crdirent_comp_fn: OK\n");
    }
    return resp_p->status;
}

static PINT_sm_action remove_check_error_code(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    char buf[64] = {0};

    PVFS_strerror_r(js_p->error_code, buf, 64);
    gossip_debug(GOSSIP_REMOVE_DEBUG, "remove_check_error_code got %s "
                 "(%d)\n", buf, js_p->error_code);
    /*
      separate error codes which are ok to return to the caller at
      this point without trying to undo what we've done and retry the
      removal
    */
    switch(js_p->error_code)
    {
        case -PVFS_ENOTEMPTY:
            /* let the error pass through for the crdirent cleanup */
            break;
        case -PVFS_EINVAL:
        case -PVFS_ENOMEM:
            /* don't undo after these errors */
            sm_p->u.remove.stored_error_code = js_p->error_code;
            js_p->error_code = RETURN_STORED_ERROR_CODE;
            break;
    }
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action remove_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    gossip_debug(GOSSIP_CLIENT_DEBUG, "remove state: cleanup\n");

    sm_p->error_code = (sm_p->u.remove.stored_error_code ?
                        sm_p->u.remove.stored_error_code :
                        js_p->error_code);

    /* NOTE: acache is invalidated by remove_helper now */
    
    /* The ncache invalidate must be done from this function, because the 
     * remove_helper may not  have all the information needed
     */
    PINT_ncache_invalidate((const char*) sm_p->u.remove.object_name,
                           (const PVFS_object_ref*) &(sm_p->parent_ref));

    PINT_msgpairarray_destroy(&sm_p->msgarray_op);
    PINT_SM_GETATTR_STATE_CLEAR(sm_p->getattr);

    PINT_SET_OP_COMPLETE;
    return SM_ACTION_TERMINATE;
}

static PINT_sm_action remove_generic_timer(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -PVFS_EINVAL;
    job_id_t tmp_id;

    gossip_debug(GOSSIP_CLIENT_DEBUG, "remove state: generic_timer\n");

    ret = job_req_sched_post_timer(
        sm_p->msgarray_op.params.retry_delay, smcb, 0, js_p, &tmp_id,
        pint_client_sm_context);

    return ret;
}

static PINT_sm_action remove_crdirent_retry_or_fail(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    gossip_debug(GOSSIP_CLIENT_DEBUG,
                 "remove state: crdirent_retry_or_fail\n");

    /* try again (up to a point) if we get a comm. failure. */
    if ((PVFS_ERROR_CLASS(-js_p->error_code) == PVFS_ERROR_BMI) &&
        (sm_p->u.remove.retry_count < sm_p->msgarray_op.params.retry_limit))
    {
        sm_p->u.remove.retry_count++;
        js_p->error_code = CRDIRENT_RETRY;
        return SM_ACTION_COMPLETE;
    }

    if ((js_p->error_code == -PVFS_EEXIST) &&
        (sm_p->u.remove.retry_count > 0))
    {
        /* assume everything worked out ok and we got the right
         * directory entry back.  there was just a transient network
         * problem along the way
         */
        js_p->error_code = 0;
        return SM_ACTION_COMPLETE;
    }

    sm_p->u.remove.stored_error_code = js_p->error_code;

    gossip_err("Error: failed to replace directory during remove recovery: entry %s for object %llu.\n",
        sm_p->u.remove.object_name,
        llu(sm_p->object_ref.handle));
    PVFS_perror_gossip("crdirent", js_p->error_code);

    PRINT_REMOVE_WARNING();
    return SM_ACTION_COMPLETE;
}


static PINT_sm_action remove_rmdirent_retry_or_fail(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    gossip_debug(GOSSIP_CLIENT_DEBUG,
                 "remove state: rmdirent_retry_or_fail\n");

    /* try again (up to a point) if we get a comm. failure. */
    if ((PVFS_ERROR_CLASS(-js_p->error_code) == PVFS_ERROR_BMI) &&
        (sm_p->u.remove.retry_count < sm_p->msgarray_op.params.retry_limit))
    {
        sm_p->u.remove.retry_count++;
        js_p->error_code = RMDIRENT_RETRY;
        return SM_ACTION_COMPLETE;
    }

    if ((js_p->error_code == -PVFS_ENOENT) &&
        (sm_p->u.remove.retry_count > 0))
    {
        /* this is a tricky error case.  Server reports ENOENT, but
         * this is not the first time we attempted the rmdirent.  It
         * may be the case that it is reporting ENOENT because one of
         * the earlier retries possibly completed.  We will treat this
         * as success, but put out an error message.  This could
         * strand objects, or remove non-empty directories, for
         * example.
         */
        gossip_err("Warning: Received ENOENT on retry to remove entry %s.\n",
            sm_p->u.remove.object_name);

        PRINT_REMOVE_WARNING();
        js_p->error_code = -PVFS_ENOENT;
        return SM_ACTION_COMPLETE;
    }

    /* other errors are preserved and passed along to the next state */
    return SM_ACTION_COMPLETE;
}

/*
 * Local variables:
 *  mode: c
 *  c-indent-level: 4
 *  c-basic-offset: 4
 * End:
 *
 * vim: ft=c ts=8 sts=4 sw=4 expandtab
 */