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

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

/** \file
 *  \ingroup sysint
 *
 *  PVFS2 system interface routines for obtaining attributes of an object
 *  (file or 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-util.h"
#include "pvfs2-util.h"
#include "pint-cached-config.h"
#include "PINT-reqproto-encode.h"
#include "pvfs2-internal.h"
#include "pvfs2-types-debug.h"
#include "dist-dir-utils.h"

/* pvfs2_client_getattr_sm
 *
 * The sm_p->msgpair structure is used to get the attributes of the
 * object itself.  We convert the original attribute mask (in
 * sm_p->u.getattr.attrmask) to ask for datafile and distribution info
 * if the user asked for file size (PVFS_ATTR_SYS_SIZE).  This allows
 * us to obtain this information (if the object turns out to be a
 * metafile) so that we can later look up the datafile sizes and
 * calculate the overall file size.
 *
 * The sm_p->msgpairarray is used to get datafile sizes, if it turns
 * out that we need them.  This space will also need to be freed, if
 * we grab these sizes.
 */

static struct profiler getattr_prof __attribute__((unused));

extern job_context_id pint_client_sm_context;

enum
{
    GETATTR_ACACHE_MISS = 1,
    GETATTR_NEED_DATAFILE_SIZES = 2,
    GETATTR_IO_RETRY = 3,
    GETATTR_NEED_DIRDATA_ATTRS = 4,
};

/* completion function prototypes */
static int getattr_object_getattr_comp_fn(
    void *v_p, struct PVFS_server_resp *resp_p, int index);
static int getattr_datafile_getattr_comp_fn(
    void *v_p, struct PVFS_server_resp *resp_p, int index);
static int getattr_dirdata_getattr_comp_fn(
    void *v_p, struct PVFS_server_resp *resp_p, int index);

%%

nested machine pvfs2_client_datafile_getattr_sizes_sm
{
    state datafile_getattr_setup_msgpairarray
    {
        run getattr_datafile_getattr_setup_msgpairarray;
        success => datafile_getattr_xfer_msgpairarray;
        default => datafile_getattr_cleanup;
    }
    
    state datafile_getattr_xfer_msgpairarray
    {
        jump pvfs2_msgpairarray_sm;
        default => datafile_getattr_retry;
    }

    state datafile_getattr_retry
    {
        run getattr_datafile_getattr_retry;
        GETATTR_IO_RETRY => datafile_getattr_xfer_msgpairarray;
        default => datafile_getattr_cleanup;
    }

    state datafile_getattr_cleanup
    {
        run getattr_datafile_getattr_cleanup;
        default => return;
    }
}


nested machine pvfs2_client_dirdata_getattr_sm
{
    state dirdata_getattr_setup_msgpairarray
    {
        run getattr_dirdata_getattr_setup_msgpairarray;
        success => dirdata_getattr_xfer_msgpairarray;
        default => dirdata_getattr_cleanup;
    }
    
    state dirdata_getattr_xfer_msgpairarray
    {
        jump pvfs2_msgpairarray_sm;
        default => dirdata_getattr_cleanup;
    }

    state dirdata_getattr_cleanup
    {
        run getattr_dirdata_getattr_cleanup;
        default => return;
    }
}


nested machine pvfs2_client_getattr_sm
{
    state acache_lookup
    {
        run getattr_acache_lookup;
        GETATTR_ACACHE_MISS => object_getattr_setup_msgpair;
        GETATTR_NEED_DATAFILE_SIZES => datafile_get_sizes;
        default => cleanup;
    }

    state object_getattr_setup_msgpair
    {
        run getattr_object_getattr_setup_msgpair;
        success => object_getattr_xfer_msgpair;
        default => cleanup;
    }

    state object_getattr_xfer_msgpair
    {
        jump pvfs2_msgpairarray_sm;
        success => acache_insert;
        GETATTR_NEED_DATAFILE_SIZES => datafile_get_sizes;
        GETATTR_NEED_DIRDATA_ATTRS => dirdata_get_attrs;
        default => object_getattr_failure;
    }

    state acache_insert
    {
        run getattr_acache_insert;
        default => cleanup;
    }

    state object_getattr_failure
    {
        run getattr_object_getattr_failure;
        default => cleanup;
    }

    state datafile_get_sizes
    {
        jump pvfs2_client_datafile_getattr_sizes_sm;
        success => acache_insert;
        default => cleanup;
    }

    state dirdata_get_attrs
    {
        jump pvfs2_client_dirdata_getattr_sm;
        success => acache_insert;
        default => cleanup;
    }

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

machine pvfs2_client_sysint_getattr_sm
{
    state dowork
    {
        jump pvfs2_client_getattr_sm;
        default => set_sys_response; 
    }

    state set_sys_response
    {
        run getattr_set_sys_response;
        default => terminate;
    }
}
    
%%


/** Initiate retrieval of object attributes.
 */
PVFS_error PVFS_isys_getattr(
    PVFS_object_ref ref,
    uint32_t attrmask, 
    const PVFS_credentials *credentials,
    PVFS_sysresp_getattr *resp_p,
    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_getattr entered\n");

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

    PINT_smcb_alloc(&smcb, PVFS_SYS_GETATTR, 
            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->error_code = 0;
    sm_p->object_ref = ref;
    sm_p->u.getattr.getattr_resp_p = resp_p;
    PVFS_hint_copy(hints, &sm_p->hints);
    PVFS_hint_add(&sm_p->hints, PVFS_HINT_HANDLE_NAME, sizeof(PVFS_handle)
                 ,&ref.handle);
    
    PINT_SM_GETATTR_STATE_FILL(
        sm_p->getattr,
        ref,
        PVFS_util_sys_to_object_attr_mask(
            attrmask),
        PVFS_TYPE_NONE,
        0);

    return PINT_client_state_machine_post(
        smcb,  op_id, user_ptr);
}

/** Retrieve object attributes.
 */
PVFS_error PVFS_sys_getattr(
    PVFS_object_ref ref,
    uint32_t attrmask, 
    const PVFS_credentials *credentials,
    PVFS_sysresp_getattr *resp_p,
    PVFS_hint hints)
{
    PVFS_error ret, error;
    PVFS_sys_op_id op_id;

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

    ret = PVFS_isys_getattr(ref, attrmask, credentials,
                            resp_p, &op_id, hints, NULL);
    if (ret)
    {
        PVFS_perror_gossip("PVFS_isys_getattr call", ret);
        return ret;
    }

    if(op_id != -1)
    {
        /* did not complete immediately, so we wait */

        ret = PVFS_sys_wait(op_id, "getattr", &error);
        if (ret)
        {
            PVFS_perror_gossip("PVFS_sys_wait call", ret);
        }
        if(error)
        {
            ret = error;
        }
        PINT_sys_release(op_id);
    }

    return ret;
}


/**
 * getattr_acache_lookup
 * @ingroup client_sm_getattr
 * 
 * This function is invoked as the first state action of the
 * getattr-dowork state machine.  It performs a lookup into the
 * attribute cache for the attribute in question, and returns
 * result codes for a hit or a miss.  
 * 
 * @param smcb This must be a valid client state machine handle, with
 * the @ref getattr field containing valid values for the
 * fsid/handle of the desired attribute (in object_ref), as well as the 
 * requested attribute mask (req_attrmask).
 *
 * @param js_p Contains the return code to be set by this function in the
 * @ref error_code field.  This determines the next state action to jump
 * to in the getattr-dowork state machine.  Possible values for js_p->error_code
 * are:
 * <ul>
 * <li><b>GETATTR_ACACHE_MISS</b> - The requested attribute was not found
 * in the attribute cache, or the attribute was found, but the attribute's
 * mask did not include values required by the requested mask (req_attrmask).
 * </li>
 * <li><b>GETATTR_NEED_DATAFILE_SIZES</b> - The requested attribute was found
 * in the attribute cache and the mask was sufficient, but the data file size
 * was requested for this handle, so we need to get that next.
 * </li>
 * <li><b>default</b> - The requested attribute was found in the attribute
 * cache and its mask values were sufficient.
 * </li>
 *
 * @return This function should always return 1 unless an error occurred
 * within the internals of the state machine.
 */
static PINT_sm_action getattr_acache_lookup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    uint32_t trimmed_mask = 0;
    int missing_attrs;
    PVFS_object_ref object_ref;
    int ret = -1;
    int attr_status = -1;
    int size_status = -1;

    js_p->error_code = 0;

    object_ref = sm_p->getattr.object_ref;

    assert(object_ref.handle != PVFS_HANDLE_NULL);
    assert(object_ref.fs_id != PVFS_FS_ID_NULL);

    gossip_debug(GOSSIP_ACACHE_DEBUG, "%s: handle %llu fsid %d\n",
      __func__, llu(object_ref.handle), object_ref.fs_id);

   
    /* The sys attr mask request is converted to object
     * attr mask values for comparison with the cached
     */
    if(sm_p->getattr.req_attrmask & PVFS_ATTR_DATA_SIZE)
    {
        sm_p->getattr.req_attrmask |= PVFS_ATTR_META_ALL;
    }

    /* if need dir timestamp or dirent_count, need dist_dir_struct */
    if(sm_p->getattr.req_attrmask & 
            (PVFS_ATTR_DIR_DIRENT_COUNT | 
             PVFS_ATTR_COMMON_ATIME | 
             PVFS_ATTR_COMMON_CTIME | 
             PVFS_ATTR_COMMON_MTIME | 
             0 )
      )
    {
        sm_p->getattr.req_attrmask |= PVFS_ATTR_DIR_DISTDIR_ATTR;
    }

    if(sm_p->getattr.flags & PINT_SM_GETATTR_BYPASS_CACHE)
    {
        gossip_debug(GOSSIP_ACACHE_DEBUG, "acache: forced acache miss: "
                    " [%llu]\n",
                      llu(object_ref.handle));
        js_p->error_code = GETATTR_ACACHE_MISS;
        return SM_ACTION_COMPLETE;
    }

    ret = PINT_acache_get_cached_entry(object_ref,
        &sm_p->getattr.attr,
        &attr_status,
        &sm_p->getattr.size,
        &size_status);
    if(ret < 0 || attr_status < 0)
    {
        gossip_debug(GOSSIP_ACACHE_DEBUG, "acache: clean acache miss: "
                    " [%llu]\n",
                      llu(object_ref.handle));
  
        js_p->error_code = GETATTR_ACACHE_MISS;
        return SM_ACTION_COMPLETE;
    }

    /* acache hit, check results */
  
    /* The sys attr mask request is converted to object
     * attr mask values for comparison with the cached
     * entry
     */
    trimmed_mask = sm_p->getattr.req_attrmask;

    gossip_debug(GOSSIP_GETATTR_DEBUG,"request attrmask:\n");
    PINT_attrmask_print(GOSSIP_GETATTR_DEBUG,sm_p->getattr.req_attrmask);

    gossip_debug(GOSSIP_GETATTR_DEBUG,"trimmed attrmask:\n");
    PINT_attrmask_print(GOSSIP_GETATTR_DEBUG,trimmed_mask);

    gossip_debug(GOSSIP_GETATTR_DEBUG,"returned attrmask:\n");
    PINT_attrmask_print(GOSSIP_GETATTR_DEBUG,sm_p->getattr.attr.mask);

    /* the trimmed mask is used for making sure that we're only
     * checking attr bits that make sense for the object type
     * since the caller may have requested all attributes in
     * the case where it doesn't know what type of object we're
     * doing the getattr against.
     */
    if (sm_p->getattr.attr.objtype == PVFS_TYPE_METAFILE)
    {
        trimmed_mask &= (PVFS_ATTR_META_ALL |
                         PVFS_ATTR_META_UNSTUFFED |
                         PVFS_ATTR_DATA_SIZE |
                         PVFS_ATTR_COMMON_ALL);
    }
    else if (sm_p->getattr.attr.objtype == PVFS_TYPE_SYMLINK)
    {
        trimmed_mask &= (PVFS_ATTR_SYMLNK_ALL | PVFS_ATTR_COMMON_ALL);
    }
    else if (sm_p->getattr.attr.objtype == PVFS_TYPE_DIRECTORY)
    {
        trimmed_mask &= (PVFS_ATTR_COMMON_ALL | PVFS_ATTR_DIR_ALL);
    }
  
    /* trimmed_mask contains the list of attributes
     * requested for a particular object, 
     * while sm_p->getattr.attr.mask contains
     * the list of attributes cached for that object.
     * The cached attributes can be used if requested
     * is less than cached, i.e. all the attributes
     * we need are already cached.  So we need to do
     * a bitwise comparison of requested <= cached.
     *
     * xor of the two masks gives us the bits that are different,
     * and-ing that result with the requested mask gives us the
     * bits in the requested mask but not in the cached mask.
     */
    missing_attrs = ((trimmed_mask ^ sm_p->getattr.attr.mask) &
                     trimmed_mask);

    gossip_debug(GOSSIP_GETATTR_DEBUG,"missing attrmask BEFORE:\n");
    PINT_attrmask_print(GOSSIP_GETATTR_DEBUG,missing_attrs);

    if ( missing_attrs & PVFS_ATTR_META_MIRROR_DFILES )
    {
       /*Mirroring is optional, so remove mirror-dfiles*/
         missing_attrs &= ~PVFS_ATTR_META_MIRROR_DFILES;   
    }
         
    gossip_debug(GOSSIP_GETATTR_DEBUG,"missing attrmask AFTER:\n");
    PINT_attrmask_print(GOSSIP_GETATTR_DEBUG,missing_attrs);

    if((missing_attrs == PVFS_ATTR_DATA_SIZE && size_status == 0) ||
        (missing_attrs == 0))
    {
        /* nothing's missing, a hit! */
        gossip_debug(GOSSIP_ACACHE_DEBUG, "acache: acache hit "
                     "[%llu]\n", llu(object_ref.handle));
        js_p->error_code = 0;
        return SM_ACTION_COMPLETE;
    }
  
    /* check if the only thing missing is the file size, then
     * we don't need to do the object getattr operation, we only
     * need to do the datafile getattr operation, so we return
     * the GETATTR_NEED_DATAFILE_SIZES error code which will make
     * the getattr-dowork state machine jump to the datafile getattr
     * operation state. 
     */
    if(missing_attrs == PVFS_ATTR_DATA_SIZE)
    {
        if(!(sm_p->getattr.attr.mask & PVFS_ATTR_META_UNSTUFFED))
        {
            /* We are missing the size, and we don't know for sure if the
             * file has been unstuffed.  In this case, act as though we
             * missed on all atributes so that we can get fresh stuffed size
             * or datafile information as needed.
             */
        }
        else
        {
            /* if the file size is requested but the distribution info
             * isn't and it hasn't been cached, then we need to
             * get that first.
             */

            js_p->error_code = GETATTR_NEED_DATAFILE_SIZES;
            gossip_debug(GOSSIP_ACACHE_DEBUG, "acache: acache hit, need sizes"
                         "[%llu]\n", llu(object_ref.handle));
            return SM_ACTION_COMPLETE;
        }
    }

    /* we missed */
    /* clean out the attributes we got from the cache; this will be
     * overwritten when we request updated information from the server
     */
    PINT_free_object_attr(&sm_p->getattr.attr);
    gossip_debug(GOSSIP_ACACHE_DEBUG, "acache: acache miss due to mask: "
        " [%llu]\n",
          llu(object_ref.handle));

    js_p->error_code = GETATTR_ACACHE_MISS;
    return SM_ACTION_COMPLETE;
}


static PINT_sm_action getattr_object_getattr_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;
    PVFS_object_ref object_ref;
    PINT_sm_msgpair_state *msg_p;

    gossip_debug(GOSSIP_CLIENT_DEBUG, "(%p) %s\n", sm_p, __func__);

    js_p->error_code = 0;

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

    object_ref = sm_p->getattr.object_ref;

    assert(object_ref.fs_id != PVFS_FS_ID_NULL);
    assert(object_ref.handle != PVFS_HANDLE_NULL);

    /* setup the msgpair to do a getattr operation */
    PINT_SERVREQ_GETATTR_FILL(
        msg_p->req,
        *sm_p->cred_p,
        object_ref.fs_id,
        object_ref.handle,
        sm_p->getattr.req_attrmask,
        sm_p->hints);
    
    msg_p->fs_id = object_ref.fs_id;
    msg_p->handle = object_ref.handle;
    msg_p->retry_flag = PVFS_MSGPAIR_RETRY;
    msg_p->comp_fn = getattr_object_getattr_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;
}

/*
  copies data from getattr response into the user supplied sys_attr
  structure.  returns 0 for directories and symlinks, and
  GETATTR_NEED_DATAFILE_SIZES for a metafile (when appropriate)
*/
static int getattr_object_getattr_comp_fn(
    void *v_p,
    struct PVFS_server_resp *resp_p,
    int index)
{
    PVFS_object_attr *attr = NULL;
    PINT_smcb *smcb = v_p;
    PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_MSGPAIR_PARENT_SM);

    assert(resp_p->op == PVFS_SERV_GETATTR);

    gossip_debug(GOSSIP_GETATTR_DEBUG,
                 "getattr_object_getattr_comp_fn called\n");

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

    /*
     * If we've reached the callback for the getattr msgpair tranfer,
     * then we can make a copy of the retrieved attribute for later
     * caching.
     */
    PINT_copy_object_attr(&sm_p->getattr.attr,
                          &resp_p->u.getattr.attr);

    attr =  &sm_p->getattr.attr;

    /* if the ref_type mask is set to a non-zero value (!PVFS_TYPE_NONE)
     * a -PVFS_error will be triggered if the
     * attributes received are not one of the the types specified.
     * This is useful so that the client can know (in some cases) that it
     * can avoid issuing an operation to the server since the server will 
     * just pass an error back anyway.
     */
    if(sm_p->getattr.ref_type &&
       sm_p->getattr.ref_type != attr->objtype)
    {
        int ret;
        gossip_debug(GOSSIP_CLIENT_DEBUG, "*** "
                     "getattr_comp_fn: Object type mismatch.\n Possibly "
                     "saving network roundtrip by returning an error\n");

        if (sm_p->getattr.ref_type == PVFS_TYPE_DIRECTORY)
        {
            ret = -PVFS_ENOTDIR;
        }
        else
        {
            assert(sm_p->getattr.ref_type == PVFS_TYPE_METAFILE);
            ret = ((attr->objtype == PVFS_TYPE_DIRECTORY) ?
                   -PVFS_EISDIR : -PVFS_EBADF);
        }
        PVFS_perror_gossip("Object Type mismatch error", ret);
        return ret;
    }

    /* do assertion checking of getattr response values, and
     * check if file sizes are needed.  With NDEBUG defined, this block
     * only checks if file sizes are needed.
     */
    switch (attr->objtype)
    {
        case PVFS_TYPE_METAFILE:
            if (sm_p->msgarray_op.msgpair.req.u.getattr.attrmask &
                PVFS_ATTR_META_DIST)
            {
                /* if we requested distribution attrs, did the distribution 
                 * get set and is the size valid?
                 */
                assert(attr->mask & PVFS_ATTR_META_DIST);
                assert(attr->u.meta.dist && (attr->u.meta.dist_size > 0));
            }
            if (sm_p->msgarray_op.msgpair.req.u.getattr.attrmask &
                PVFS_ATTR_META_MIRROR_DFILES)
            {
                if (attr->mask & PVFS_ATTR_META_MIRROR_DFILES)
                {    assert(attr->u.meta.mirror_dfile_array &&
                          (attr->u.meta.mirror_copies_count > 0));
                     gossip_debug(GOSSIP_GETATTR_DEBUG,"%s: Mirror handles and "
                                                       "copy count retrieved.\n"
                                                     ,__func__);
                }
                else
                {
                   gossip_debug(GOSSIP_GETATTR_DEBUG,"%s: request attribute "
                   "mask says to get the mirror dfiles, if they exist, \nbut "
                   "the response attribute mask says they were not retrieved. "
                   "This is okay.\n"
                   ,__func__);
                }
            }
            if (sm_p->msgarray_op.msgpair.req.u.getattr.attrmask &
                PVFS_ATTR_META_DFILES)
            {
                /* if we requested the datafile handles for the file, did
                 * the datafile array get populated?
                 */
                assert(attr->u.meta.dfile_array &&
                       (attr->u.meta.dfile_count > 0));

                gossip_debug(GOSSIP_GETATTR_DEBUG,
                             "getattr_object_getattr_comp_fn: "
                             "%d datafiles.\n", attr->u.meta.dfile_count);
                
                /* if we need the size, that should be the only time we're 
                 * going to have to do a full data file fetch. 
                 * (that's expensive)
                 */
                if (sm_p->getattr.req_attrmask & PVFS_ATTR_DATA_SIZE)
                {
                    /* is the file stuffed? */
                    if(!(attr->mask & PVFS_ATTR_META_UNSTUFFED))
                    {
                        /* we can compute the size without doing any more
                         * getattr requests
                         */
                        gossip_debug(GOSSIP_GETATTR_DEBUG,
                            "getattr_object_getattr_comp_fn: "
                            "detected stuffed file.\n");
                        return(0);
                    }
                    /* if caller asked for the size, then we need
                     * to jump to the datafile_getattr state, which
                     * will retrieve the datafile sizes for us.
                     */
                    return GETATTR_NEED_DATAFILE_SIZES;
                }
            }
            return 0;
        case PVFS_TYPE_DIRECTORY:
        {
            uint32_t dirent_file_count_i;
            int tmp_i;
            unsigned char *c;

            /* dirent_count will be collected later */
            gossip_debug(GOSSIP_CLIENT_DEBUG,
                "getattr comp_fn [%p] "
                "dfile_count = %d "
                "dist_name_len = %d "
                "dist_params_len = %d\n ",
                attr,
                attr->u.dir.hint.dfile_count,
                attr->u.dir.hint.dist_name_len,
                attr->u.dir.hint.dist_params_len);

            if(attr->mask & PVFS_ATTR_DIR_DISTDIR_ATTR)
            {

                gossip_debug(GOSSIP_CLIENT_DEBUG, 
                        "sys-getattr: dist-dir-attr "
                        "with tree_height=%d, num_servers=%d, bitmap_size=%d, "
                        "split_size=%d, server_no=%d and branch_level=%d\n",
                        attr->u.dir.dist_dir_attr.tree_height,
                        attr->u.dir.dist_dir_attr.num_servers,
                        attr->u.dir.dist_dir_attr.bitmap_size,
                        attr->u.dir.dist_dir_attr.split_size,
                        attr->u.dir.dist_dir_attr.server_no,
                        attr->u.dir.dist_dir_attr.branch_level);

                /* gossip bitmap */
                gossip_debug(GOSSIP_CLIENT_DEBUG,
                        "sys-getattr: dist_dir_bitmap as:\n");
                for(tmp_i = attr->u.dir.dist_dir_attr.bitmap_size - 1;
                        tmp_i >= 0 ; tmp_i--)
                {
                    c = (unsigned char *)(attr->u.dir.dist_dir_bitmap + tmp_i);
                    gossip_debug(GOSSIP_MKDIR_DEBUG,
                            " i=%d : %02x %02x %02x %02x\n",
                            tmp_i, c[3], c[2], c[1], c[0]);
                }
                gossip_debug(GOSSIP_CLIENT_DEBUG, "\n");

                /* gossip dirdata handles */
                gossip_debug(GOSSIP_CLIENT_DEBUG,
                        "sys-getattr: dirdata_handles as:\n");
                for (dirent_file_count_i = 0;
                        dirent_file_count_i < 
                        attr->u.dir.dist_dir_attr.num_servers;
                        dirent_file_count_i++) 
                {
                    gossip_debug(GOSSIP_CLIENT_DEBUG,
                            "    dirdata_handle[%d] = %llu\n",
                            dirent_file_count_i,
                            llu(attr->u.dir.dirdata_handles[dirent_file_count_i]));
                }
            }

            /* need to gather dirdata attr */
            if(sm_p->getattr.req_attrmask & 
                    (PVFS_ATTR_DIR_DIRENT_COUNT | 
                     PVFS_ATTR_COMMON_ATIME | 
                     PVFS_ATTR_COMMON_CTIME | 
                     PVFS_ATTR_COMMON_MTIME | 
                     0 )
              )
            {
                assert(attr->mask & PVFS_ATTR_DIR_DISTDIR_ATTR);

                gossip_debug(GOSSIP_CLIENT_DEBUG,
                        "  !!! need to contact dirdata to gather dirent_count and timestamps!\n");

                return GETATTR_NEED_DIRDATA_ATTRS;

            }

            return 0;
        }
        case PVFS_TYPE_SYMLINK:
            return 0;
        case PVFS_TYPE_DATAFILE:
            return 0;
        case PVFS_TYPE_DIRDATA:
            return 0;
        case PVFS_TYPE_INTERNAL:
            return 0;
        default:
            gossip_err("error: getattr_object_getattr_comp_fn: "
                       "handle refers to invalid object type\n");
    }
    return -PVFS_EINVAL;
}


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

    gossip_debug(GOSSIP_CLIENT_DEBUG
                ,"(%p) getattr state: getattr_object_getattr_failure\n"
                ,sm_p);

    if ((js_p->error_code != -PVFS_ENOENT) &&
        (js_p->error_code != -PVFS_EINVAL))
    {
        PVFS_perror_gossip("getattr_object_getattr_failure ",
                           js_p->error_code);
    }

    return SM_ACTION_COMPLETE;
}

/* NOTE:  This nested state machine allocates and stores the results in getattr.size_array.  So,
 * if you call this state machine directly, do not allocate space prior to calling it to avoid a
 * nasty memory leak.
*/
static PINT_sm_action getattr_datafile_getattr_setup_msgpairarray(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PINT_sm_getattr_state *getattr = &(sm_p->getattr);
    int ret = -PVFS_EINVAL;
    int i = 0;
    PVFS_object_attr *attr = NULL;
    PINT_sm_msgpair_state *msg_p = NULL;
    uint64_t mirror_retry = (sm_p->getattr.attr.mask & PVFS_ATTR_META_MIRROR_DFILES);

    gossip_debug(GOSSIP_MIRROR_DEBUG,"Executing getattr_datafile_getattr_setup_msgpairarray...\n");
    gossip_debug(GOSSIP_MIRROR_DEBUG,"%s: Are we mirroring? %s\n",__func__
                                    ,(mirror_retry ? "YES" : "NO")); 
    gossip_debug(GOSSIP_MIRROR_DEBUG,"%s: attr.mask:0x%08x \tmirror_retry:0x%08x\n"
                ,__func__
                ,sm_p->getattr.attr.mask
                ,(unsigned int)mirror_retry);

    js_p->error_code = 0;

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

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

    /*initialize the size_array, which will hold the size of each datahandle*/
    PINT_SM_DATAFILE_SIZE_ARRAY_INIT(&getattr->size_array,attr->u.meta.dfile_count);

    /* initialize mir_ctx_array: one context for each handle in the file */
    getattr->mir_ctx_array = malloc(attr->u.meta.dfile_count *
                                    sizeof(*getattr->mir_ctx_array));
    if (!getattr->mir_ctx_array)
    {
        gossip_lerr("Unable to allocate mirror context array.\n");
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    memset(getattr->mir_ctx_array,0,attr->u.meta.dfile_count * 
                                   sizeof(*getattr->mir_ctx_array));
    getattr->mir_ctx_count = attr->u.meta.dfile_count;

    for (i=0; i<getattr->mir_ctx_count; i++)
    {
       getattr->mir_ctx_array[i].original_datahandle = attr->u.meta.dfile_array[i];
       getattr->mir_ctx_array[i].original_server_nr = i;
    }

    /* allocate handle array and populate */
    PVFS_handle *handles = malloc(sizeof(PVFS_handle) * attr->u.meta.dfile_count);
    if (!handles)
    {
        gossip_lerr("Unable to allocation local handles array.\n");
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    memset(handles,0,sizeof(PVFS_handle) * attr->u.meta.dfile_count);
    memcpy(handles,attr->u.meta.dfile_array,attr->u.meta.dfile_count *
                                            sizeof(PVFS_handle));

    /*allocate index-to-server array and populate*/
    getattr->index_to_server = malloc(sizeof(uint32_t)*attr->u.meta.dfile_count);
    if (!getattr->index_to_server)
    {
        gossip_lerr("Unable to allocate index-to-server array.\n");
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    memset(getattr->index_to_server,0,sizeof(uint32_t)*attr->u.meta.dfile_count);
    for (i=0; i<attr->u.meta.dfile_count; i++)
    {
        getattr->index_to_server[i] = (uint32_t)i;
    }

//START_PROFILER(getattr_prof);
    PINT_SERVREQ_TREE_GET_FILE_SIZE_FILL(
            msg_p->req,
            *sm_p->cred_p,
            sm_p->getattr.object_ref.fs_id,
            0,
            attr->u.meta.dfile_count,
            handles,
            (mirror_retry ? 1 : 0),
            sm_p->hints);

    msg_p->fs_id = sm_p->getattr.object_ref.fs_id;
    msg_p->handle = handles[0];
    msg_p->comp_fn = getattr_datafile_getattr_comp_fn;

    ret = PINT_cached_config_map_to_server(
        &msg_p->svr_addr, msg_p->handle, msg_p->fs_id);
    if (ret)
    {
        gossip_lerr("Unable to map server address for this handle(%llu) and "
                    "filesystem(%d)\n"
                   ,llu(msg_p->handle)
                   ,msg_p->fs_id);
        js_p->error_code = ret;
        return SM_ACTION_COMPLETE;
    }

    /* set retry flag based on mirroring option...if mirroring, we will handle
     * retries from this machine; if not, msgpairarray will handle retries.
    */
    if (mirror_retry)
    {
        msg_p->retry_flag = PVFS_MSGPAIR_NO_RETRY;
    }
    else
    {
        msg_p->retry_flag = PVFS_MSGPAIR_RETRY;
    }


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

static int getattr_datafile_getattr_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);
    PINT_sm_msgpair_state *msg = &(sm_p->msgarray_op.msgarray[index]);
    struct PVFS_servreq_tree_get_file_size *tree = 
                                              &(msg->req.u.tree_get_file_size);

    PINT_sm_getattr_state *getattr = &(sm_p->getattr);
    PINT_client_getattr_mirror_ctx *ctx = NULL;
    uint32_t server_nr = 0;
    int i = 0;

    if (resp_p->status)
    {   /* tree request had a problem */
        return resp_p->status;
    }

    assert(resp_p->op == PVFS_SERV_TREE_GET_FILE_SIZE);


    /* if we are mirroring, then we need to check the error code returned from
     * each server. If an error is found, mirroring will try to get the size 
     * from a different server.  Below, we are marking which handles completed 
     * successfully, which tells mirroring NOT to retry them.
    */
    if ( getattr->attr.mask & PVFS_ATTR_META_MIRROR_DFILES)  
    {
       for (i=0; i<resp_p->u.tree_get_file_size.handle_count; i++)
       {
          if (resp_p->u.tree_get_file_size.error[i] != 0)
          {   /* error retrieving size for this handle..we will retry it. */
             continue;
          }
          server_nr = getattr->index_to_server[i];
          sm_p->getattr.size_array[server_nr] = resp_p->u.tree_get_file_size.size[i];
          ctx = &(getattr->mir_ctx_array[server_nr]);
          ctx->msg_completed = 1; 

          /*For completed messages, update the size array with the file size
           *just retrieved.
          */
          getattr->size_array[server_nr] = resp_p->u.tree_get_file_size.size[i];
          gossip_debug(GOSSIP_GETATTR_DEBUG,"%s: size[%d]:%lld \thandle:%llu\n"
                                         ,__func__
                                         ,i
                                         ,llu(getattr->size_array[i])
                                         ,llu(tree->handle_array[i]));
       }/*end for*/
    }
    else
    {
     /* if we are NOT mirroring and an error is found for an individual handle, 
      * then we must invalidate the size array and return an error code.
     */
       for (i=0; i<resp_p->u.tree_get_file_size.handle_count; i++)
       {
          gossip_debug(GOSSIP_GETATTR_DEBUG,"%s: error[%d]:%d"
                                               "\tsize[%d]:%d\n"
                                           ,__func__
                                           ,i
                                           ,(int)resp_p->u.tree_get_file_size.error[i]
                                           ,i
                                           ,(int)resp_p->u.tree_get_file_size.size[i]);

          if (resp_p->u.tree_get_file_size.error[i] != 0)
          {
              gossip_debug(GOSSIP_GETATTR_DEBUG,"%s: error[%d] is %d\n"
                                               ,__func__
                                               ,i
                                               ,resp_p->u.tree_get_file_size.error[i]);
              memset(getattr->size_array,0,sizeof(*getattr->size_array));
              return (resp_p->u.tree_get_file_size.error[i]);
          }

          getattr->size_array[i] = resp_p->u.tree_get_file_size.size[i];
       }/*end for*/

    }/*end if*/

    return(0);
}/*end getattr_datafile_getattr_comp_fn*/

static PINT_sm_action getattr_datafile_getattr_retry(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm   *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    struct PVFS_object_attr *attr = &(sm_p->getattr.attr);
    PVFS_metafile_attr *meta = &(attr->u.meta);
    PINT_sm_getattr_state *getattr = &(sm_p->getattr);
    PINT_client_getattr_mirror_ctx *ctx = NULL;
    PINT_sm_msgarray_op *mop = &(sm_p->msgarray_op);
    PINT_sm_msgpair_state *msg = &(mop->msgarray[0]);
    char *enc_req_bytes = NULL;
    struct PVFS_servreq_tree_get_file_size *tree = 
                                              &(msg->req.u.tree_get_file_size);
    uint32_t retry_msg_count = 0;
    uint32_t index = 0;
    uint32_t copies = 0;
    uint32_t server_nr = 0;
    int i   = 0;
    int j   = 0;
    int k   = 0;
    int ret = 0;

    gossip_debug(GOSSIP_CLIENT_DEBUG, 
                 "(%p) getattr state: "
                 "getattr_datafile_getattr_retry\n", sm_p);

    /*We only need to retry if we have mirrors; otherwise, msgpairarray 
     *has already handled the retries.
    */
    if (!(attr->mask & PVFS_ATTR_META_MIRROR_DFILES))  
    {  
       /*we are NOT mirroring.*/
       return SM_ACTION_COMPLETE;
    }
    
    /* How many handles need to be retried? */
    for (i=0; i<tree->num_data_files; i++)
    {
        server_nr = getattr->index_to_server[i];
        ctx = &(getattr->mir_ctx_array[server_nr]);
        if (ctx->msg_completed == 0)
            retry_msg_count++;
    }

    /* no retries needed */
    if (retry_msg_count == 0)
    {
        return SM_ACTION_COMPLETE;
    }

    /* do we have any retries available? */
    if (getattr->retry_count >= mop->params.retry_limit)
    {
        /* at this point, we have msgpairs that need to be retried, but we
         * we have met our retry limit.  so, we must invalidate the size array,
         * since we don't have all of the necessary sizes AND return an error.
        */
        memset(getattr->size_array,0,sizeof(*getattr->size_array) * getattr->size);
        js_p->error_code = (js_p->error_code ? js_p->error_code : -PVFS_ETIME);
        gossip_err("%s: Ran out of retries(%d)\n",__func__
                                                 ,getattr->retry_count);
        return SM_ACTION_COMPLETE;
    }

    /*allocate temporary index-to-server array*/
    uint32_t *tmp_server_nr = malloc(sizeof(uint32_t) * retry_msg_count);
    if (!tmp_server_nr)
    {
        gossip_lerr("Unable to allocate temporary index-to-server array.\n");
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    memset(tmp_server_nr,0,sizeof(uint32_t) * retry_msg_count);

    /*allocate temporary handle array*/
    PVFS_handle *tmp_handles = malloc(sizeof(PVFS_handle) * retry_msg_count);
    if (!tmp_handles)
    {
        gossip_lerr("Unable to allocate temporary handle array.\n");
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    memset(tmp_handles,0,sizeof(PVFS_handle) * retry_msg_count);

    /* okay. let's setup new handles to retry. 
    */
    for (i=0,j=0; i<tree->num_data_files; i++)
    {
        server_nr = getattr->index_to_server[i];
        ctx = &(getattr->mir_ctx_array[server_nr]);
     
        /* don't process completed messages */
        if (ctx->msg_completed)
           continue;

        /* for incomplete messages, cleanup memory, if necessary */ 
        enc_req_bytes = (char *)&(msg->encoded_req);
        for (k=0; k<sizeof(msg->encoded_req); k++)
        {
           if (enc_req_bytes[k] != '\0')
           {
              PINT_encode_release(&(msg->encoded_req),PINT_ENCODE_REQ);
              break;
           }
        }/*end for*/

        if (msg->encoded_resp_p)
        {
            BMI_memfree(msg->svr_addr
                       ,msg->encoded_resp_p
                       ,msg->max_resp_sz
                       ,BMI_RECV);
        }
        
        /* Should we use the original datahandle? */
        if (ctx->retry_original)
        {
            ctx->retry_original = 0;
            tmp_handles[j]   = ctx->original_datahandle;
            tmp_server_nr[j] = ctx->original_server_nr;
            j++;
            continue;
        }/*end retry_original*/

        /* otherwise, get next mirrored handle.  note:  if a mirrored handle is
         * zero, then this means that the creation of this mirrored object 
         * failed for its particular server.  in this case, get the next valid 
         * handle.  as a last resort, retry the original handle.
        */
        copies = ctx->current_copies_count;
        for (;copies < meta->mirror_copies_count; copies++)
        {
            index = (copies*meta->dfile_count) + server_nr;
            if (meta->mirror_dfile_array[index] != 0)
            {  /* we have found a valid mirrored handle */
               tmp_handles[j]   = meta->mirror_dfile_array[index];
               tmp_server_nr[j] = server_nr;
               j++;
               break;
            }
        }

        /* if we haven't found a valid mirrored handle, retry the original
         * datahandle.
        */
        if ( copies == meta->mirror_copies_count )
        {
           tmp_handles[j]   = ctx->original_datahandle;
           tmp_server_nr[j] = ctx->original_server_nr;
           j++;
           ctx->retry_original = 0;
           ctx->current_copies_count = 0;
           getattr->retry_count++;
           continue;
        }/*end if we have to use the original*/

        /* otherwise, setup for the next retry event */
        ctx->current_copies_count++;
        if (ctx->current_copies_count == meta->mirror_copies_count)
        {
           ctx->current_copies_count = 0;
           ctx->retry_original = 1;
           getattr->retry_count++;
        }
    }/*end for each handle in the old request*/

    /*replace values in old tree request*/
    free(tree->handle_array);
    tree->handle_array = tmp_handles;
    tree->num_data_files = retry_msg_count;
    
    /*replace values in old message request*/
    msg->handle = tmp_handles[0];
    msg->svr_addr=0;
    ret = PINT_cached_config_map_to_server(&msg->svr_addr
                                          ,msg->handle
                                          ,msg->fs_id);
    if (ret)
    {
        gossip_lerr("Unable to determine server address for handle(%llu) and "
                    "file system(%d).\n"
                   ,llu(msg->handle)
                   ,msg->fs_id);
        js_p->error_code = -PVFS_EINVAL;
        return SM_ACTION_COMPLETE;
    }

    /*save index-to-server array*/
    free(getattr->index_to_server);
    getattr->index_to_server = tmp_server_nr;

    /* Push the msgarray_op and jump to msgpairarray.sm */
    PINT_sm_push_frame(smcb,0,mop);
    js_p->error_code=GETATTR_IO_RETRY;

    return SM_ACTION_COMPLETE;
} /*end datafile_getattr_retry*/






static PINT_sm_action getattr_datafile_getattr_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
//FINISH_PROFILER("getattr", getattr_prof, 1);
    PINT_sm_getattr_state *getattr = &(sm_p->getattr);

    /* cleanup handle_array created for the one tree request */
    PINT_sm_msgpair_state *msg_p = &(sm_p->msgarray_op.msgarray[0]);
    if (msg_p->req.u.tree_get_file_size.handle_array)
       free(msg_p->req.u.tree_get_file_size.handle_array);
    
    /* cleanup tree request */
    PINT_msgpairarray_destroy(&sm_p->msgarray_op);

    /* cleanup memory that may have been used for mirrored retries.*/
    if (getattr->mir_ctx_array)
       free(getattr->mir_ctx_array);
    if (getattr->index_to_server)
       free(getattr->index_to_server);


    return SM_ACTION_COMPLETE;
}


static PINT_sm_action getattr_dirdata_getattr_setup_msgpairarray(
        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;
    int i = 0;
    PVFS_object_attr *attr = NULL;
    PINT_sm_msgpair_state *msg_p = NULL;
    int tmp_index, cur_index;

    js_p->error_code = 0;

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

    assert(attr->mask & PVFS_ATTR_DIR_DISTDIR_ATTR);

    /* initialize size_array, used to store the dirent_count of every dirdata handle */
    PINT_SM_DATAFILE_SIZE_ARRAY_INIT(&sm_p->getattr.size_array,
            attr->u.dir.dist_dir_attr.num_servers);

    sm_p->getattr.active_dirdata_index = (int *)malloc(
            sizeof(*sm_p->getattr.active_dirdata_index) *
            attr->u.dir.dist_dir_attr.num_servers);

    if(!sm_p->getattr.active_dirdata_index)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }


    /* find out active dirdata handles */
    tmp_index = 0;
    for(i=0; i<attr->u.dir.dist_dir_attr.num_servers; i++)
    {
        if(PINT_is_dist_dir_bucket_active(&attr->u.dir.dist_dir_attr,
                    attr->u.dir.dist_dir_bitmap, i))
        {
            sm_p->getattr.active_dirdata_index[tmp_index] = i;
            tmp_index++;
        }
    }
    assert(tmp_index > 0);

    /* initialize msgpair array */
    ret = PINT_msgpairarray_init(&sm_p->msgarray_op, 
            tmp_index);
    if(ret != 0)
    {
        js_p->error_code = ret;
        return SM_ACTION_COMPLETE;
    }

    /* prepare to post the getattr send/recv pairs for all dirdata*/
    foreach_msgpair(&sm_p->msgarray_op, msg_p, i)
    {
        cur_index = sm_p->getattr.active_dirdata_index[i];

        gossip_debug(GOSSIP_CLIENT_DEBUG,
                     "getattr: posting dirdata getattr[%d] ([%d]:%lld,%d)\n",
                     i, cur_index,
                     llu(attr->u.dir.dirdata_handles[cur_index]),
                     sm_p->getattr.object_ref.fs_id);

        PINT_SERVREQ_GETATTR_FILL(
            msg_p->req,
            *sm_p->cred_p,
            sm_p->getattr.object_ref.fs_id,
            attr->u.dir.dirdata_handles[cur_index],
            PVFS_ATTR_COMMON_ALL|PVFS_ATTR_DIR_DIRENT_COUNT, /* only interested in timestamp and dirent_count */
            sm_p->hints);

        /* fill in msgpair structure components */
        msg_p->fs_id = sm_p->getattr.object_ref.fs_id;
        msg_p->handle = attr->u.dir.dirdata_handles[cur_index];
        msg_p->retry_flag = PVFS_MSGPAIR_RETRY;
        msg_p->comp_fn = getattr_dirdata_getattr_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;
            break;
        }
    }

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

}


static int getattr_dirdata_getattr_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);
    PVFS_object_attr *attr_p = NULL;
    PVFS_object_attr *resp_attr_p = NULL;
    int cur_index;

    gossip_debug(GOSSIP_GETATTR_DEBUG,
                 "getattr_dirdata_getattr_comp_fn called\n");
    gossip_debug(GOSSIP_CLIENT_DEBUG, "dirdata getattr[%d] got response %d\n",
                 index, resp_p->status);

    if (resp_p->status)
    {   
        return resp_p->status;
    }

    assert(resp_p->op == PVFS_SERV_GETATTR);

    attr_p = &sm_p->getattr.attr;
    resp_attr_p = &resp_p->u.getattr.attr;
    cur_index = sm_p->getattr.active_dirdata_index[index];

    assert(resp_attr_p->objtype == PVFS_TYPE_DIRDATA);
    /* also assert timestamp? */
    assert(resp_attr_p->mask & PVFS_ATTR_DIR_DIRENT_COUNT);
    assert( (cur_index >= 0) && 
            (cur_index < attr_p->u.dir.dist_dir_attr.num_servers));

    gossip_debug(GOSSIP_CLIENT_DEBUG, "dirdata getattr[%d] (#[%d]) returns attrs: "
            "[dirent_count=%llu, atime = %llu, mtime = %llu, ctime = %llu]\n",
            index, cur_index,
            llu(resp_attr_p->u.dir.dirent_count),
            llu(resp_attr_p->atime),
            llu(resp_attr_p->mtime),
            llu(resp_attr_p->ctime));

    /* update timestamp and dirent_count */
    sm_p->getattr.size_array[cur_index] = resp_attr_p->u.dir.dirent_count; 
    attr_p->u.dir.dirent_count += resp_attr_p->u.dir.dirent_count;
    if(attr_p->atime < resp_attr_p->atime)
    {
        attr_p->atime = resp_attr_p->atime;
    }
    if(attr_p->ctime < resp_attr_p->ctime)
    {
        attr_p->ctime = resp_attr_p->ctime;
    }
    if(attr_p->mtime < resp_attr_p->mtime)
    {
        attr_p->mtime = resp_attr_p->mtime;
    }

    gossip_debug(GOSSIP_CLIENT_DEBUG, "updated directory attrs: "
            "[dirent_count=%llu, atime = %llu, mtime = %llu, ctime = %llu]\n",
            llu(attr_p->u.dir.dirent_count),
            llu(attr_p->atime),
            llu(attr_p->mtime),
            llu(attr_p->ctime));

    return 0;
}


static PINT_sm_action getattr_dirdata_getattr_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    
    /* cleanup tree request */
    PINT_msgpairarray_destroy(&sm_p->msgarray_op);

    if(sm_p->getattr.active_dirdata_index)
    {
        free(sm_p->getattr.active_dirdata_index);
    }

    return SM_ACTION_COMPLETE;
}




static PINT_sm_action getattr_acache_insert(
        struct PINT_smcb *smcb, job_status_s *js_p)
{   
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_size* tmp_size = NULL;

    if( sm_p->getattr.attr.objtype == PVFS_TYPE_METAFILE || 
        sm_p->getattr.attr.objtype == PVFS_TYPE_DIRECTORY ||
        sm_p->getattr.attr.objtype == PVFS_TYPE_SYMLINK )
    {
        /* see if we have a size value to cache */
        if (sm_p->getattr.attr.objtype == PVFS_TYPE_METAFILE &&
            sm_p->getattr.req_attrmask & PVFS_ATTR_DATA_SIZE)
        {                                           
            if(!(sm_p->getattr.attr.mask & PVFS_ATTR_META_UNSTUFFED))
            {
                /* stuffed file case */
                sm_p->getattr.size = sm_p->getattr.attr.u.meta.stuffed_size;
                tmp_size = &sm_p->getattr.size;
                gossip_debug(GOSSIP_GETATTR_DEBUG, "getattr_acache_insert "
                  "calculated stuffed logical size of %lld\n", lld(*tmp_size));
            }
            else
            {
                /* compute size as requested */
                assert(sm_p->getattr.attr.u.meta.dist);
                assert(sm_p->getattr.attr.u.meta.dist->methods &&
                       sm_p->getattr.attr.u.meta.dist->methods->logical_file_size);

                sm_p->getattr.size =
                    (sm_p->getattr.attr.u.meta.dist->methods->logical_file_size)(
                    sm_p->getattr.attr.u.meta.dist->params,
                    sm_p->getattr.attr.u.meta.dfile_count,
                    sm_p->getattr.size_array);

                tmp_size = &sm_p->getattr.size;
                gossip_debug(GOSSIP_GETATTR_DEBUG,"getattr_acache_insert calculated"
                                                  " unstuffed logical size of %lld\n"
                                                 , lld(*tmp_size));
            }
        }

        PINT_acache_update(sm_p->getattr.object_ref,
            &sm_p->getattr.attr,
            tmp_size);

        gossip_debug(GOSSIP_CLIENT_DEBUG, "trying to add object "
                     "reference to acache\n");
    }

    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PINT_sm_getattr_state *getattr = &(sm_p->getattr);

    gossip_debug(GOSSIP_CLIENT_DEBUG,
                 "(%p) getattr state: getattr_cleanup\n", sm_p);

    gossip_debug(GOSSIP_GETATTR_DEBUG
                ,"%s: js_p->error_code:%d \tgetattr->attr.mask:0x%08x\n"
                ,__func__
                ,js_p->error_code
                ,getattr->attr.mask);

    sm_p->error_code = js_p->error_code;

    /* cleanup size array if not asked to keep; is allocated if datafile sizes or dirent_count are retrieved */
    /* keep_size_array is initialized outside the state machine, now only used in sys-readdir.sm */
    if (!getattr->keep_size_array && getattr->size_array)
    {
        gossip_debug(GOSSIP_CLIENT_DEBUG,
                 "getattr_cleanup: freeint getattr->size_array. \n");
        free(getattr->size_array);
        getattr->size_array = NULL;
    }

    /* cleanup getattr when an error occurs */
    if (js_p->error_code)
    {
      if (getattr->attr.mask & PVFS_ATTR_META_DFILES)
      {
         if (getattr->attr.u.meta.dfile_array)
            free(getattr->attr.u.meta.dfile_array);
      }

      if (getattr->attr.mask & PVFS_ATTR_META_MIRROR_DFILES)
      {
         if (getattr->attr.u.meta.mirror_dfile_array)
            free(getattr->attr.u.meta.mirror_dfile_array);
      }

      if (getattr->attr.mask & PVFS_ATTR_META_DIST)
      {
         PINT_dist_free(getattr->attr.u.meta.dist);
      }
    }/*end if error*/

    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_set_sys_response(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_client_sm *sm_p = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_sysresp_getattr * sysresp = NULL;
    PVFS_object_attr *attr = NULL;

    if(js_p->error_code != 0)
    {
        PINT_SET_OP_COMPLETE;
        return SM_ACTION_TERMINATE;
    }

    attr = &sm_p->getattr.attr;
    assert(attr);
    
    /* If we get to this state action, 
     * the getattr state machine was invoked, so
     * we can assume that one of the PVFS_[i]sys_getattr functions
     * was called, and the response field must be filled in for the
     * user.
     */

    sysresp = sm_p->u.getattr.getattr_resp_p;
    
    /*
     * if we retrieved a symlink target, copy it for the caller; this
     * target path will be handed all the way back up to the caller via
     * the PVFS_sys_attr object.  The caller of PVFS_[i]sys_getattr
     * must free it.
     */
    if(attr->objtype == PVFS_TYPE_SYMLINK &&
       attr->mask & PVFS_ATTR_SYMLNK_TARGET)
    {
        assert(attr->u.sym.target_path_len > 0);
        assert(attr->u.sym.target_path);

        sysresp->attr.link_target = strdup(attr->u.sym.target_path);
        if (!sysresp->attr.link_target)
        {
           js_p->error_code = -PVFS_ENOMEM;
           PINT_SET_OP_COMPLETE;
           return SM_ACTION_TERMINATE;
        }
    }

    if(attr->objtype == PVFS_TYPE_METAFILE) 
    {
       /* Copy if there are any special object specific flags */
       sysresp->attr.flags = attr->u.meta.hint.flags;
       /* special case for when users ask for dfile count */
       if (sm_p->getattr.req_attrmask & PVFS_ATTR_META_DFILES)
       {
           sysresp->attr.dfile_count = attr->u.meta.dfile_count;
       }
       if (sm_p->getattr.req_attrmask & PVFS_ATTR_META_MIRROR_DFILES)
       {
         sysresp->attr.mirror_copies_count = attr->u.meta.mirror_copies_count;
       }
    }
    if (attr->objtype == PVFS_TYPE_DIRECTORY)
    {
        gossip_debug(GOSSIP_CLIENT_DEBUG, "dfile_count: %d\n", 
            attr->u.dir.hint.dfile_count);
        gossip_debug(GOSSIP_CLIENT_DEBUG, "dist_name_len = %d, "
            "dist_params_len = %d\n",
            attr->u.dir.hint.dist_name_len, attr->u.dir.hint.dist_params_len);
        sysresp->attr.dfile_count = attr->u.dir.hint.dfile_count;


       if (sm_p->getattr.req_attrmask & PVFS_ATTR_DIR_DISTDIR_ATTR)
       {
           sysresp->attr.dirdata_count = 
               attr->u.dir.dist_dir_attr.num_servers;
       }

        /* 
         * If we retrieved any extended attributes for the directory
         * in question, the caller's responsibility to free it up
         */
        if (attr->u.dir.hint.dist_name_len > 0 && 
            (sm_p->getattr.req_attrmask & PVFS_ATTR_DIR_HINT))
        {
            sysresp->attr.dist_name = strdup(attr->u.dir.hint.dist_name);
            if (!sysresp->attr.dist_name)
            {
                js_p->error_code = -PVFS_ENOMEM;
                PINT_SET_OP_COMPLETE;
                return SM_ACTION_TERMINATE;
            }
            gossip_debug(GOSSIP_CLIENT_DEBUG, "dist_name_hint: %s\n"
                                            , sysresp->attr.dist_name);
        }
        if (attr->u.dir.hint.dist_params_len > 0 &&
            (sm_p->getattr.req_attrmask & PVFS_ATTR_DIR_HINT))
        {
            sysresp->attr.dist_params = strdup(attr->u.dir.hint.dist_params);
            if (!sysresp->attr.dist_params)
            {
                free(sysresp->attr.dist_name);
                sysresp->attr.dist_name = NULL;
                js_p->error_code = -PVFS_ENOMEM;
                PINT_SET_OP_COMPLETE;
                return SM_ACTION_TERMINATE;
            }
            gossip_debug(GOSSIP_CLIENT_DEBUG, "dist_name_params: %s\n"
                                            , sysresp->attr.dist_params);
        }
    }

    /* copy outgoing sys_attr fields from returned object_attr */
    sysresp->attr.owner = attr->owner;
    sysresp->attr.group = attr->group;
    sysresp->attr.perms = attr->perms;
    sysresp->attr.atime = attr->atime;
    sysresp->attr.mtime = attr->mtime;
    sysresp->attr.ctime = attr->ctime;
    sysresp->attr.mask  = PVFS_util_object_to_sys_attr_mask(attr->mask);
    sysresp->attr.size  = 0;
    sysresp->attr.objtype = attr->objtype;

    if (js_p->error_code == 0)
    {
        /* convert outgoing attribute mask based on what we got */
        sysresp->attr.mask = PVFS_util_object_to_sys_attr_mask(
            sm_p->getattr.attr.mask);

       if (sm_p->getattr.req_attrmask & PVFS_ATTR_DATA_SIZE)
        {
            if( attr->objtype == PVFS_TYPE_DATAFILE )
            {
                sysresp->attr.size = attr->u.data.size;
            }
            else
            {
                sysresp->attr.size = sm_p->getattr.size;
            }

            sysresp->attr.mask |= PVFS_ATTR_SYS_SIZE;
        }

        if(attr->mask & PVFS_ATTR_META_DIST)
        {
            /* we have enough information to set a block size */
            sysresp->attr.blksize = attr->u.meta.dist->methods->get_blksize(
                attr->u.meta.dist->params);
            sysresp->attr.mask |= PVFS_ATTR_SYS_BLKSIZE;
        }

        /* if this is a symlink, add the link target */
        if (sm_p->getattr.req_attrmask & PVFS_ATTR_SYMLNK_TARGET)
        {
            sysresp->attr.mask |= PVFS_ATTR_SYS_LNK_TARGET;
        }

        if(sm_p->getattr.req_attrmask & PVFS_ATTR_DIR_DIRENT_COUNT)
        {
            sysresp->attr.dirent_count = attr->u.dir.dirent_count;
            sysresp->attr.mask |= PVFS_ATTR_SYS_DIRENT_COUNT;
        }

        if (sm_p->getattr.req_attrmask & PVFS_ATTR_DIR_DISTDIR_ATTR)
        {
/* TODO: Do we need this? */
        }
    }
    else
    {
        /* in case of failure, blank out response */ 
        memset(sm_p->u.getattr.getattr_resp_p,
               0, sizeof(PVFS_sysresp_getattr));
    }

    PINT_SM_GETATTR_STATE_CLEAR(sm_p->getattr);

    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
 */