root/branches/Orange-Elaine-Distr-Dir-Branch/src/client/sysint/sys-readdir.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 reading entries from a directory.
 */
#include <string.h>
#include <assert.h>

#include "client-state-machine.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 "pint-util.h"
#include "pvfs2-internal.h"

extern job_context_id pint_client_sm_context;

static int readdir_msg_comp_fn(
    void *v_p, struct PVFS_server_resp *resp_p, int index);

%%

nested machine pvfs2_client_readdir_sm
{
    state init
    {
        run readdir_init;
        default => readdir_getattr;
    }

    state readdir_getattr
    {
        jump pvfs2_client_getattr_sm;
        success => readdir_msg_setup_msgpair;
        default => cleanup;
    }

    state readdir_msg_setup_msgpair
    {
        run readdir_msg_setup_msgpair;
        success => readdir_msg_xfer_msgpair;
        default => readdir_msg_failure;
    }

    state readdir_msg_xfer_msgpair
    {
        jump pvfs2_msgpairarray_sm;
        success => cleanup;
        default => readdir_msg_failure;
    }

    state readdir_msg_failure
    {
        run readdir_msg_failure;
        default => cleanup;
    }

    state cleanup
    {
        run readdir_cleanup;
        default => return;
    }
}

machine pvfs2_client_sysint_readdir_sm
{
    state dowork
    {
        jump pvfs2_client_readdir_sm;
        default => do_cleanup;
    }
    state do_cleanup
    {
        run do_cleanup;
        default => terminate;
    }
}

%%

/** Initiate reading of entries from a directory.
 *
 *  \param token opaque value used to track position in directory
 *         when more than one read is required.
 *  \param pvfs_dirent_incount maximum number of entries to read, if
 *         available, starting from token.
 */
PVFS_error PVFS_isys_readdir(
    PVFS_object_ref ref,
    PVFS_ds_position token, 
    int32_t pvfs_dirent_incount,
    const PVFS_credentials *credentials,
    PVFS_sysresp_readdir *resp,
    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_readdir entered\n");

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

    if (pvfs_dirent_incount > PVFS_REQ_LIMIT_DIRENT_COUNT)
    {
        gossip_lerr("PVFS_isys_readdir unable to handle request "
                    "for %d entries.\n", pvfs_dirent_incount);
        return ret;
    }

    PINT_smcb_alloc(&smcb, PVFS_SYS_READDIR,
             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, ref.fs_id);
    PINT_init_sysint_credentials(sm_p->cred_p, credentials);
    sm_p->u.readdir.readdir_resp = resp;
    sm_p->object_ref = ref;
    PVFS_hint_copy(hints, &sm_p->hints);
    PVFS_hint_add(&sm_p->hints, PVFS_HINT_HANDLE_NAME, sizeof(PVFS_handle), &ref.handle);

    /* point the sm dirent array and outcount to the readdir response field */
    sm_p->readdir.dirent_array = &resp->dirent_array;
    sm_p->readdir.dirent_outcount = &resp->pvfs_dirent_outcount;
    sm_p->readdir.token = &resp->token;
    sm_p->readdir.directory_version = &resp->directory_version;

    sm_p->readdir.pos_token = sm_p->u.readdir.pos_token = token;
    sm_p->readdir.dirent_limit = sm_p->u.readdir.dirent_limit = pvfs_dirent_incount;

    gossip_debug(GOSSIP_READDIR_DEBUG, "Doing readdir on handle "
                 "%llu on fs %d\n", llu(ref.handle), ref.fs_id);

    return PINT_client_state_machine_post(
        smcb,  op_id, user_ptr);
}

/** Read entries from a directory.
 *
 *  \param token opaque value used to track position in directory
 *         when more than one read is required.
 *  \param pvfs_dirent_incount maximum number of entries to read, if
 *         available, starting from token.
 */
PVFS_error PVFS_sys_readdir(
    PVFS_object_ref ref,
    PVFS_ds_position token, 
    int32_t pvfs_dirent_incount,
    const PVFS_credentials *credentials,
    PVFS_sysresp_readdir *resp,
    PVFS_hint hints)
{
    PVFS_error ret = -PVFS_EINVAL, error = 0;
    PVFS_sys_op_id op_id;

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

    ret = PVFS_isys_readdir(ref, token, pvfs_dirent_incount,
                            credentials, resp, &op_id, hints, NULL);
    if (ret)
    {
        PVFS_perror_gossip("PVFS_isys_readdir call", ret);
        error = ret;
    }
    else
    {
        ret = PVFS_sys_wait(op_id, "readdir", &error);
        if (ret)
        {
            PVFS_perror_gossip("PVFS_sys_wait call", ret);
            error = ret;
        }
    }

    PINT_sys_release(op_id);
    return error;
}

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

static PINT_sm_action readdir_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, "readdir state: init\n");

    PINT_SM_GETATTR_STATE_FILL(
        sm_p->getattr,
        sm_p->object_ref,
        (PVFS_ATTR_DIR_ALL|PVFS_ATTR_DIR_DIRENT_FILES),
        PVFS_TYPE_DIRECTORY,
        0);
    
    assert(js_p->error_code == 0);

    return SM_ACTION_COMPLETE;
}

static PINT_sm_action readdir_msg_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;

    gossip_debug(GOSSIP_CLIENT_DEBUG, "readdir state: "
                 "readdir_msg_setup_msgpair\n");

    if (js_p->error_code)
    {
        return SM_ACTION_COMPLETE;
    }
    js_p->error_code = 0;

    gossip_debug(GOSSIP_READDIR_DEBUG," readdir: posting readdir req\n");

    /* TODO: Need to find the correct dirent_handle */
    gossip_debug(
        GOSSIP_READDIR_DEBUG, "%llu|%llu|%d | token is %llu | limit is %d\n",
        llu(sm_p->object_ref.handle),
        llu(sm_p->getattr.attr.u.dir.dirent_handle[0]),
        sm_p->object_ref.fs_id,
        llu(sm_p->readdir.pos_token),
        sm_p->readdir.dirent_limit);

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

    PINT_SERVREQ_READDIR_FILL(
        msg_p->req,
        *sm_p->cred_p,
        sm_p->object_ref.fs_id,
        sm_p->object_ref.handle,
        sm_p->getattr.attr.u.dir.dirent_handle[0],
        sm_p->u.readdir.pos_token,
        sm_p->u.readdir.dirent_limit,
        sm_p->hints);

    msg_p->fs_id = sm_p->object_ref.fs_id;
    msg_p->handle = sm_p->object_ref.handle;
    msg_p->retry_flag = PVFS_MSGPAIR_RETRY;
    msg_p->comp_fn = readdir_msg_comp_fn;

    ret = PINT_cached_config_map_to_server(
        &msg_p->svr_addr, sm_p->object_ref.handle,
        sm_p->object_ref.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 readdir_msg_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);
    
    gossip_debug(GOSSIP_CLIENT_DEBUG, "readdir_msg_comp_fn\n");

    assert(resp_p->op == PVFS_SERV_READDIR);

    if (resp_p->status != 0)
    {
        return resp_p->status;
    }

    /* convert servresp_readdir response to a sysresp_readdir obj */

    *(sm_p->readdir.token) = resp_p->u.readdir.token;
    *(sm_p->readdir.directory_version) =
        resp_p->u.readdir.directory_version;
    *(sm_p->readdir.dirent_outcount) =
        resp_p->u.readdir.dirent_count;
    if (*(sm_p->readdir.dirent_outcount) > 0)
    {
        int dirent_array_len =
            (sizeof(PVFS_dirent) * *(sm_p->readdir.dirent_outcount));

        /* this dirent_array MUST be freed by caller */
        *(sm_p->readdir.dirent_array) =
            (PVFS_dirent *) malloc(dirent_array_len);
        assert(*(sm_p->readdir.dirent_array));

        memcpy(*(sm_p->readdir.dirent_array),
               resp_p->u.readdir.dirent_array, dirent_array_len);
    }

    gossip_debug(GOSSIP_READDIR_DEBUG, "*** Got %d directory entries "
                 "[version %lld]\n",
                 *(sm_p->readdir.dirent_outcount),
                 lld(*(sm_p->readdir.directory_version)));

    return 0;
}

static PINT_sm_action readdir_msg_failure(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    gossip_debug(GOSSIP_CLIENT_DEBUG,
                 "readdir state: readdir_msg_failure\n");
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action readdir_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int i = 0;
    PVFS_object_ref tmp_ref;
    gossip_debug(GOSSIP_CLIENT_DEBUG, "readdir state: cleanup\n");

    if(js_p->error_code == 0)
    {
        /* insert all handles into the ncache while we have them */
        tmp_ref.fs_id = sm_p->object_ref.fs_id;
        for(i = 0; i < *(sm_p->readdir.dirent_outcount); i++)
        {
            tmp_ref.handle = (*(sm_p->readdir.dirent_array))[i].handle;
            PINT_ncache_update(
                (const char *) (*(sm_p->readdir.dirent_array))[i].d_name,
                (const PVFS_object_ref *) &(tmp_ref),
                (const PVFS_object_ref *) &(sm_p->object_ref));
        }
    }

    PINT_SM_GETATTR_STATE_CLEAR(sm_p->getattr);

    if(js_p->error_code != 0)
    {
        PINT_acache_invalidate(sm_p->object_ref);
    }
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action do_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, "readdir state: do_cleanup\n");

    PINT_free_object_attr(&sm_p->getattr.attr);
    sm_p->error_code = js_p->error_code;
    gossip_debug(GOSSIP_READDIR_DEBUG, " final return code is %d\n",
                 sm_p->error_code);

    PINT_SET_OP_COMPLETE;
    return SM_ACTION_TERMINATE;
}



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