root/branches/orange-next/src/server/get-attr.sm @ 8950

/* 
 * (C) 2001 Clemson University and The University of Chicago 
 *
 * See COPYING in top-level directory.
 *
 * Changes by Acxiom Corporation to add dirent_count field to attributes
 * Copyright © Acxiom Corporation, 2005.
 */

/* pvfs2_get_attr_sm
 *
 * This state machine handles incoming server getattr operations.  These
 * are the operations sent by PVFS_sys_getattr() among others.
 *
 * The pvfs2_prelude_sm is responsible for reading the actual metadata
 * to begin with, because it does this as part of the permission checking
 * process.
 */

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

#include "server-config.h"
#include "pvfs2-server.h"
#include "pvfs2-attr.h"
#include "pvfs2-types.h"
#include "pvfs2-types-debug.h"
#include "pvfs2-util.h"
#include "pint-util.h"
#include "pvfs2-internal.h"
#include "pint-cached-config.h"
#include "pvfs2-mirror.h"

PINT_server_trove_keys_s Trove_Special_Keys[] =
{
    {"user.pvfs2.dist_name"    , SPECIAL_DIST_NAME_KEYLEN},
    {"user.pvfs2.dist_params"  , SPECIAL_DIST_PARAMS_KEYLEN},
    {"user.pvfs2.num_dfiles"   , SPECIAL_NUM_DFILES_KEYLEN},
    {"user.pvfs2.meta_hint"    , SPECIAL_METAFILE_HINT_KEYLEN},
    {"user.pvfs2.mirror.copies", SPECIAL_MIRROR_COPIES_KEYLEN},
    {"user.pvfs2.mirror.handles", SPECIAL_MIRROR_HANDLES_KEYLEN},
    {"user.pvfs2.mirror.status" , SPECIAL_MIRROR_STATUS_KEYLEN},
};

enum
{
    STATE_METAFILE  = 7,
    STATE_SYMLINK   = 9,
    STATE_DIR       = 10,
    STATE_DIR_HINT  = 11,
    STATE_DONE      = 12,
    SKIP_NEXT_STATE = 13,
};

static void free_nested_getattr_data(struct PINT_server_op *s_op);

%%

nested machine pvfs2_get_attr_work_sm
{
    state verify_attribs
    {
        run getattr_verify_attribs;
        STATE_SYMLINK => read_symlink_target;
        STATE_METAFILE => read_metafile_hint;
        STATE_DIR => get_dirdata_handle;
        default => setup_resp;
    }

    state read_symlink_target
    {
        run getattr_read_symlink_target;
        default => setup_resp;
    }

    state read_metafile_hint
    {
        run getattr_read_metafile_hint;
        default => interpret_metafile_hint;
    }

    state interpret_metafile_hint
    {
        run getattr_interpret_metafile_hint;
        STATE_METAFILE => read_metafile_datafile_handles_if_required;
        default => setup_resp;
    }

    state read_metafile_datafile_handles_if_required
    {
        run getattr_read_metafile_datafile_handles_if_required;
        success => datafile_handles_safety_check;
        default => setup_resp;
    }

    state datafile_handles_safety_check
    {
        run getattr_datafile_handles_safety_check;
        success => read_mirrored_copies_count_if_required; 
        default => setup_resp;
    }

    state read_mirrored_copies_count_if_required
    {
        run getattr_read_mirrored_copies_count_if_required;
        SKIP_NEXT_STATE => read_metafile_distribution_if_required;
        default => read_mirrored_handles_if_required;
    }

    state read_mirrored_handles_if_required
    {
        run getattr_read_mirrored_handles_if_required;
        SKIP_NEXT_STATE => read_metafile_distribution_if_required;
        default => mirrored_handles_safety_check;
    }
 
    state mirrored_handles_safety_check
    {
        run getattr_mirrored_handles_safety_check;
        success => read_metafile_distribution_if_required;
        default => setup_resp;
    }

    state read_metafile_distribution_if_required
    {
        run getattr_read_metafile_distribution_if_required;
        default => interpret_metafile_distribution;
    }

    state interpret_metafile_distribution
    {
        run interpret_metafile_distribution;
        success => detect_stuffed;
        default => setup_resp;
    }

    state detect_stuffed
    {
        run getattr_detect_stuffed;
        default => read_stuffed_size;
    }

    state read_stuffed_size
    {
        run getattr_read_stuffed_size;
        success => interpret_stuffed_size;
        default => setup_resp;
    }

    state interpret_stuffed_size
    {
        run getattr_interpret_stuffed_size;
        default => setup_resp;
    }

    state get_dirdata_handle
    {
        run getattr_get_dirdata_handle;
        success => get_dirent_count;
        default => setup_resp;
    }

    state get_dirent_count
    {
        run getattr_get_dirent_count;
        STATE_DIR_HINT => get_dir_hint;
        default => interpret_dirent_count;
    }

    state interpret_dirent_count
    {
        run getattr_interpret_dirent_count;
        default => get_dir_hint;
    }

    state get_dir_hint
    {
        run getattr_get_dir_hint;
        STATE_DONE => setup_resp;
        default => interpret_dir_hint;
    }

    state interpret_dir_hint
    {
        run getattr_interpret_dir_hint;
        default => setup_resp;
    }

    state setup_resp
    {
        run getattr_setup_resp;
        default => return;
    }
}

nested machine pvfs2_get_attr_with_prelude_sm
{
    state init
    {
        run getattr_with_prelude_init;
        default => prelude;
    }

    state prelude
    {
        jump pvfs2_prelude_sm;
        success => setup_op;
        default => return;
    }

    state setup_op
    {
        run getattr_setup_op;
        default => do_work;
    }

    state do_work
    {
        jump pvfs2_get_attr_work_sm;
        default => return;
    }
}

machine pvfs2_get_attr_sm
{
    state work
    {
        jump pvfs2_get_attr_with_prelude_sm;
        default => final_response;
    }

    state final_response
    {
        jump pvfs2_final_response_sm;
        default => cleanup;
    }

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

%%



/* getattr_verify_attribs()
 *
 * We initialize the attribute mask that will be returned in this
 * function.  This mask can be augmented in some of the other states.
 */
static PINT_sm_action getattr_verify_attribs(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_object_attr *resp_attr = NULL;

    js_p->error_code = 0;

    /*
      explicitly copy basic attributes structure (read in from the
      prelude.sm for the matching dspace) into response to be sent
      back to the client.  this is mostly for readability here to be
      sure we know which fields are valid in the response at this
      point.
    */
    resp_attr = &s_op->resp.u.getattr.attr;
    memset(resp_attr, 0, sizeof(PVFS_object_attr));

    resp_attr->owner = s_op->attr.owner;
    resp_attr->group = s_op->attr.group;
    resp_attr->perms = s_op->attr.perms;
    resp_attr->atime = s_op->attr.atime;

    resp_attr->mtime = PINT_util_mkversion_time(s_op->attr.mtime);
    if (resp_attr->mtime == 0)
    {
        /*
          this is a compatibility hack to allow existing storage
          spaces to be automagically converted to this versioned time
          on-disk format slowly over time and doing the right thing in
          the meantime
        */
        resp_attr->mtime = s_op->attr.mtime;

        gossip_debug(GOSSIP_GETATTR_DEBUG, " No version found!  Using "
                     "mtime %llu\n", llu(resp_attr->mtime));
    }
    else
    {
        gossip_debug(
            GOSSIP_GETATTR_DEBUG, " VERSION is %llu, mtime is %llu\n",
            llu(s_op->attr.mtime), llu(resp_attr->mtime));
    }

    resp_attr->ctime = s_op->attr.ctime;
    resp_attr->mask = s_op->attr.mask;
    resp_attr->objtype = s_op->attr.objtype;
    resp_attr->u.meta.dfile_count = s_op->attr.u.meta.dfile_count;
    resp_attr->u.meta.dist_size = s_op->attr.u.meta.dist_size;

#if 0
    gossip_debug(
        GOSSIP_GETATTR_DEBUG,
        "+  _DSPACE_ retrieved attrs: [owner = %d, group = %d\n\t"
        "perms = %o, type = %d, atime = %llu, mtime = %llu\n\t"
        "ctime = %llu, dfile_count = %d, dist_size = %d]\n",
        resp_attr->owner, resp_attr->group, resp_attr->perms,
        resp_attr->objtype, llu(resp_attr->atime),
        llu(resp_attr->mtime), llu(resp_attr->ctime),
        (int)resp_attr->u.meta.dfile_count,
        (int)resp_attr->u.meta.dist_size);
#endif

    /*
      weed out the attr mask of the response based on what the client
      request asked for.  also, check if we need to retrieve more
      information before returning the response to the client (by
      guiding the state machine to get it).

      we can safely do this now that we have the type of the object
      (read in from the dspace, not stored in the resp_attr), and we
      have the original client request attr mask
      (s_op->u.getattr.attrmask).
    */
    switch(resp_attr->objtype)
    {
        case PVFS_TYPE_METAFILE:
            PINT_ACCESS_DEBUG(s_op, GOSSIP_ACCESS_DEBUG, "type: metafile\n");
            gossip_debug(GOSSIP_GETATTR_DEBUG,
                         "  Req handle %llu refers to a metafile\n",
                         llu(s_op->u.getattr.handle));

            if (s_op->u.getattr.attrmask & PVFS_ATTR_META_DFILES)
            {
                gossip_debug(GOSSIP_GETATTR_DEBUG,
                             " dspace has dfile_count of %d\n",
                             resp_attr->u.meta.dfile_count);
                resp_attr->mask |= PVFS_ATTR_META_DFILES;
            }
            else
            {
                gossip_debug(GOSSIP_GETATTR_DEBUG, " client doesn't want "
                             "dfile info, clearing response attr mask\n");
                resp_attr->mask &= ~PVFS_ATTR_META_DFILES;
            }

            if (s_op->u.getattr.attrmask & PVFS_ATTR_META_DIST)
            {
                gossip_debug(GOSSIP_GETATTR_DEBUG,
                             " dspace has dist size of %d\n",
                             resp_attr->u.meta.dist_size);

                resp_attr->mask |= PVFS_ATTR_META_DIST;
            }
            else
            {
                gossip_debug(GOSSIP_GETATTR_DEBUG, " client doesn't want "
                             "dist info, clearing response attr mask\n");

                resp_attr->mask &= ~PVFS_ATTR_META_DIST;
            }

            if (s_op->u.getattr.attrmask & PVFS_ATTR_META_MIRROR_DFILES)
            {
               gossip_debug(GOSSIP_GETATTR_DEBUG,"client wants mirrored "
                                                 "handles.\n");
               resp_attr->mask |= PVFS_ATTR_META_MIRROR_DFILES;
               resp_attr->u.meta.mirror_copies_count = 0;
               resp_attr->u.meta.mirror_dfile_array  = NULL;
            }
            else
            {
               gossip_debug(GOSSIP_GETATTR_DEBUG,"client doesn't want "
                                                 "mirrored handles.\n");
               resp_attr->mask &= ~(PVFS_ATTR_META_MIRROR_DFILES);
            }
            js_p->error_code = STATE_METAFILE;
            break;
        case PVFS_TYPE_DATAFILE:
            PINT_ACCESS_DEBUG(s_op, GOSSIP_ACCESS_DEBUG, "type: datafile\n");
            /*
              note: the prelude already retrieved the size for us, so
              there's no special action that needs to be taken if we have
              a datafile here (other than adjusting our mask to include
              the data information and copying the retrieved size from the
              ds_attribute the prelude used)
            */
            resp_attr->u.data.size = s_op->ds_attr.u.datafile.b_size;
            resp_attr->mask |= PVFS_ATTR_DATA_ALL;

            gossip_debug(GOSSIP_GETATTR_DEBUG, "  handle %llu refers to "
                         "a datafile (size = %lld).\n",
                         llu(s_op->u.getattr.handle),
                         lld(resp_attr->u.data.size));
            break;
        case PVFS_TYPE_DIRECTORY:
            PINT_ACCESS_DEBUG(s_op, GOSSIP_ACCESS_DEBUG, "type: directory\n");
            if (s_op->u.getattr.attrmask & PVFS_ATTR_DIR_DIRENT_COUNT)
            {
                gossip_debug(GOSSIP_GETATTR_DEBUG,
                             " getattr: dirent_count needed.\n");
                assert(resp_attr->mask & PVFS_ATTR_COMMON_ALL);
                resp_attr->mask |= PVFS_ATTR_DIR_DIRENT_COUNT;
                js_p->error_code = STATE_DIR;
            }
            else
            {
                gossip_debug(GOSSIP_GETATTR_DEBUG,
                             " getattr: dirent_count not needed.\n");
                js_p->error_code = 0;
                assert(resp_attr->mask & PVFS_ATTR_COMMON_ALL);
            }
            if (s_op->u.getattr.attrmask & PVFS_ATTR_DIR_HINT)
            {
                gossip_debug(GOSSIP_GETATTR_DEBUG,
                            " getattr: dfile_count needed.\n");
                assert(resp_attr->mask & PVFS_ATTR_COMMON_ALL);
                resp_attr->mask |= PVFS_ATTR_DIR_HINT;
                js_p->error_code = STATE_DIR;
            }
            else
            {
                gossip_debug(GOSSIP_GETATTR_DEBUG,
                            " getattr: dfile_count not needed\n");
                assert(resp_attr->mask & PVFS_ATTR_COMMON_ALL);
            }
            break;
        case PVFS_TYPE_DIRDATA:
            PINT_ACCESS_DEBUG(s_op, GOSSIP_ACCESS_DEBUG, "type: dirdata\n");
            gossip_debug(
                GOSSIP_GETATTR_DEBUG, "  handle %llu refers to "
                "a dirdata object. doing nothing special\n",
                llu(s_op->u.getattr.handle));
            assert(resp_attr->mask & PVFS_ATTR_COMMON_ALL);
            break;
        case PVFS_TYPE_SYMLINK:
            PINT_ACCESS_DEBUG(s_op, GOSSIP_ACCESS_DEBUG, "type: symlink\n");
            gossip_debug(
                GOSSIP_GETATTR_DEBUG, "  handle %llu refers to a symlink.\n",
                llu(s_op->u.getattr.handle));

            /*
              we'll definitely have to fetch the symlink target in this
              case, as the prelude will never retrieve it for us
            */
            js_p->error_code = STATE_SYMLINK;
            break;
        case PVFS_TYPE_INTERNAL:
            PINT_ACCESS_DEBUG(s_op, GOSSIP_ACCESS_DEBUG, "type: symlink\n");
            /* nothing interesting to add; this is meaningless to a client */
            break;
        default:
            /* if we don't understand the object type, then it probably indicates
             * a bug or some data corruption.  All trove objects should have a
             * type set.
             */
            gossip_err(
                "Error: got unknown type when verifying attributes for "
                "handle %llu.\n", 
                llu(s_op->u.getattr.handle));
            js_p->error_code = -PVFS_ENXIO;
            break;
    }

    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_read_symlink_target(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret;
    job_id_t i;

    /* if we don't need to fill in the symlink target, skip it */
    if (!(s_op->u.getattr.attrmask & PVFS_ATTR_SYMLNK_TARGET))
    {
        gossip_debug(GOSSIP_GETATTR_DEBUG, "skipping symlink target read\n");
        js_p->error_code = 0;
        return SM_ACTION_COMPLETE;
    }
    
    s_op->key.buffer    = Trove_Common_Keys[SYMLINK_TARGET_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[SYMLINK_TARGET_KEY].size;

    /*
      optimistically add mask value to indicate the symlink target is
      filled (error_code is checked in next state)
    */
    s_op->resp.u.getattr.attr.mask |= PVFS_ATTR_SYMLNK_TARGET;

    s_op->resp.u.getattr.attr.u.sym.target_path_len = PVFS_NAME_MAX;
    s_op->resp.u.getattr.attr.u.sym.target_path =
        malloc(s_op->resp.u.getattr.attr.u.sym.target_path_len);
    if (!s_op->resp.u.getattr.attr.u.sym.target_path)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }

    if(s_op->free_val)
    {
        free(s_op->val.buffer);
    }
    s_op->val.buffer = s_op->resp.u.getattr.attr.u.sym.target_path;
    s_op->val.buffer_sz = s_op->resp.u.getattr.attr.u.sym.target_path_len;
    /* this will get cleaned up with attr structure */
    s_op->free_val = 0;

    ret = job_trove_keyval_read(
        s_op->u.getattr.fs_id, s_op->u.getattr.handle,
        &s_op->key, &s_op->val,
        0,
        NULL, smcb, 0, js_p,
        &i, server_job_context, s_op->req->hints);


    return ret;
}

static PINT_sm_action getattr_interpret_metafile_hint(
    PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_object_attr *resp_attr = NULL;
    PVFS_metafile_attr *meta = &(s_op->resp.u.getattr.attr.u.meta); 
    resp_attr = &s_op->resp.u.getattr.attr;

    assert(resp_attr->objtype == PVFS_TYPE_METAFILE);

    if (js_p->error_code == 0 || js_p->error_code == -TROVE_ENOENT)
    {
        if (js_p->error_code == 0)
        {
            memcpy(&(meta->hint), s_op->val.buffer, sizeof(meta->hint));
        }
        if ((resp_attr->mask & PVFS_ATTR_META_DFILES) ||
            (resp_attr->mask & PVFS_ATTR_META_DIST)   ||
            (resp_attr->mask & PVFS_ATTR_META_MIRROR_DFILES))
        {
            gossip_debug(GOSSIP_GETATTR_DEBUG, " * client wants extra "
                         "meta info, about to retrieve it now\n");
            js_p->error_code = STATE_METAFILE;
            if ( (resp_attr->mask  & PVFS_ATTR_META_MIRROR_DFILES) && 
                !(meta->hint.flags & PVFS_MIRROR_FL) ) 
                  resp_attr->mask &= ~(PVFS_ATTR_META_MIRROR_DFILES);
        }
        else
        {
            gossip_debug(GOSSIP_GETATTR_DEBUG, " * client doesn't want "
                         "extra meta info, preparing response now\n");
            js_p->error_code = 0;
        }
    } else {
        /*If we hit an error the DIST & DFILES are no longer valid*/
        resp_attr->mask &= ~PVFS_ATTR_META_DIST;
        resp_attr->mask &= ~PVFS_ATTR_META_DFILES;
        resp_attr->mask &= ~PVFS_ATTR_META_MIRROR_DFILES;
    }
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_read_metafile_hint(
    struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -PVFS_EINVAL;
    job_id_t i;
    char *buf = NULL;

    assert(s_op->attr.objtype == PVFS_TYPE_METAFILE);
    buf = (char *) calloc(sizeof(PVFS_metafile_hint) + 1, 1);
    if (buf == NULL) 
    {
        js_p->error_code = -PVFS_ENOMEM;
        /*If we hit an error the DIST & DFILES are no longer valid*/
        s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_DIST;
        s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_DFILES;
        s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;
        return 1;
    }

    js_p->error_code = 0;

    s_op->key.buffer = Trove_Special_Keys[METAFILE_HINT_KEY].key;
    s_op->key.buffer_sz = Trove_Special_Keys[METAFILE_HINT_KEY].size;

    if(s_op->free_val)
    {
        free(s_op->val.buffer);
    }
    s_op->val.buffer = buf;
    s_op->val.buffer_sz = sizeof(s_op->resp.u.getattr.attr.u.meta.hint) + 1;
    s_op->free_val = 1;

    gossip_debug(GOSSIP_GETATTR_DEBUG,
                 "  reading metafile hint (coll_id = %d, "
                 "handle = %llu, key = %s (%d), val_buf = %p (%d))\n",
                 s_op->u.getattr.fs_id,
                 llu(s_op->u.getattr.handle), (char *)s_op->key.buffer,
                 s_op->key.buffer_sz, s_op->val.buffer,
                 s_op->val.buffer_sz);

    ret = job_trove_keyval_read(
        s_op->u.getattr.fs_id, s_op->u.getattr.handle,
        &s_op->key, &s_op->val, 
        0, 
        NULL, smcb, 0, js_p,
        &i, server_job_context, s_op->req->hints);

    return ret;
}

static PINT_sm_action getattr_read_metafile_datafile_handles_if_required(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -PVFS_EINVAL;
    int dfile_count = 0;
    job_id_t i;

    assert(s_op->attr.objtype == PVFS_TYPE_METAFILE);

    js_p->error_code = 0;

    /* if we don't need to fill in the dfiles, skip them */
    if (!(s_op->u.getattr.attrmask & PVFS_ATTR_META_DFILES))
    {
        gossip_debug(GOSSIP_GETATTR_DEBUG, "skipping data handle read\n");
        return SM_ACTION_COMPLETE;
    }

    dfile_count = s_op->resp.u.getattr.attr.u.meta.dfile_count;

    gossip_debug(GOSSIP_GETATTR_DEBUG,
                 " request has dfile_count of %d | dspace has %d\n",
                 s_op->resp.u.getattr.attr.u.meta.dfile_count,
                 s_op->resp.u.getattr.attr.u.meta.dfile_count);

    /* verify that the retrieved dfile count is sane */
    if (!PVFS_REQ_LIMIT_DFILE_COUNT_IS_VALID(dfile_count))
    {
        gossip_err("The requested dfile count of %d is invalid; "
                   "aborting operation.\n", dfile_count);
        gossip_err(
            "+ attrs read from dspace: (owner = %d, group = %d, "
            "perms = %o, type = %d\n   atime = %lld, mtime = %lld, "
            "ctime = %lld |\n   dfile_count = %d | dist_size = %d)\n",
            s_op->resp.u.getattr.attr.owner,
            s_op->resp.u.getattr.attr.group, 
            s_op->resp.u.getattr.attr.perms,
            s_op->resp.u.getattr.attr.objtype, 
            lld(s_op->resp.u.getattr.attr.atime),
            lld(s_op->resp.u.getattr.attr.mtime), 
            lld(s_op->resp.u.getattr.attr.ctime),
            (int)s_op->resp.u.getattr.attr.u.meta.dfile_count,
            (int)s_op->resp.u.getattr.attr.u.meta.dist_size);

        gossip_err("handle: %llu (%llx), fsid: %d.\n",
            llu(s_op->u.getattr.handle), llu(s_op->u.getattr.handle),
            (int)s_op->u.getattr.fs_id);

        /*If we hit an error the DIST & DFILES are no longer valid*/
        s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_DIST;
        s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_DFILES;
        s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;
        
        js_p->error_code = -PVFS_EOVERFLOW;
        return SM_ACTION_COMPLETE;
    }

    s_op->key.buffer = Trove_Common_Keys[METAFILE_HANDLES_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[METAFILE_HANDLES_KEY].size;

    /* add mask value to indicate the data file array is filled */
    s_op->resp.u.getattr.attr.mask |= PVFS_ATTR_META_DFILES;

    s_op->resp.u.getattr.attr.u.meta.dfile_array =
        malloc(dfile_count * sizeof(PVFS_handle));
    if (!s_op->resp.u.getattr.attr.u.meta.dfile_array)
    {
        gossip_err("Cannot allocate dfile array of count %d\n",
                   dfile_count);
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }

    if(s_op->free_val)
    {
        free(s_op->val.buffer);
    }
    s_op->val.buffer = s_op->resp.u.getattr.attr.u.meta.dfile_array;
    s_op->val.buffer_sz = (dfile_count * sizeof(PVFS_handle));
    /* this will get cleaned up with attr structure */
    s_op->free_val = 0;

    gossip_debug(GOSSIP_GETATTR_DEBUG,
                 "  reading %d datafile handles (coll_id = %d, "
                 "handle = %llu, key = %s (%d), val_buf = %p (%d))\n",
                 dfile_count, s_op->u.getattr.fs_id,
                 llu(s_op->u.getattr.handle), (char *)s_op->key.buffer,
                 s_op->key.buffer_sz, s_op->val.buffer,
                 s_op->val.buffer_sz);

    ret = job_trove_keyval_read(
        s_op->u.getattr.fs_id
       ,s_op->u.getattr.handle
       ,&s_op->key
       ,&s_op->val
       ,0
       ,NULL
       ,smcb
       ,0
       ,js_p
       ,&i
       ,server_job_context
       ,s_op->req->hints);

    return ret;
}


static PINT_sm_action getattr_read_mirrored_copies_count_if_required(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    gossip_debug(GOSSIP_MIRROR_DEBUG,"Executing %s...\n",__func__);

    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    struct PVFS_server_resp *resp = &(s_op->resp);
    PVFS_metafile_attr *meta =  &(resp->u.getattr.attr.u.meta);
    int ret = -PVFS_EINVAL;
    job_id_t job_id;

   /* Are we mirroring? */
    if (!(resp->u.getattr.attr.mask & PVFS_ATTR_META_MIRROR_DFILES))
    {
        gossip_debug(GOSSIP_MIRROR_DEBUG,"\tMirroring is NOT turned on "
                                         "for this handle(%llu)..\n"
                                        ,llu(s_op->u.getattr.handle));
        js_p->error_code = SKIP_NEXT_STATE;

        return SM_ACTION_COMPLETE;
    }

    gossip_debug(GOSSIP_MIRROR_DEBUG,"\tMirroring IS turned on for this "
                                     "handle(%llu)...\n"
                                    ,llu(s_op->u.getattr.handle));

    js_p->error_code = 0;

    /* setup job to read user.pvfs2.mirror.copies */

    /* initialize */
    if (s_op->free_val)
       free(s_op->val.buffer);
    memset(&(s_op->val),0,sizeof(s_op->val));
    memset(&(s_op->key),0,sizeof(s_op->key));

    /* set key = user.pvfs2.mirror.copies */
    s_op->key.buffer    = Trove_Special_Keys[MIRROR_COPIES_KEY].key;
    s_op->key.buffer_sz = Trove_Special_Keys[MIRROR_COPIES_KEY].size;

    /* setup space for retrieved value */
    meta->mirror_copies_count = 0;
    meta->mirror_dfile_array  = NULL;
    s_op->val.buffer = &(meta->mirror_copies_count);
    s_op->val.buffer_sz = sizeof(meta->mirror_copies_count);
    s_op->free_val = 0;

    /* submit job to read the value */
    ret = job_trove_keyval_read(
        s_op->u.getattr.fs_id
       ,s_op->u.getattr.handle
       ,&s_op->key
       ,&s_op->val
       ,0
       ,NULL
       ,smcb
       ,0
       ,js_p
       ,&job_id
       ,server_job_context
       ,s_op->req->hints);

    return ret;
}/*end getattr_read_mirrored_copies_if_required*/



static PINT_sm_action getattr_read_mirrored_handles_if_required(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
   gossip_debug(GOSSIP_MIRROR_DEBUG,"Executing %s ...\n",__func__);

    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    struct PVFS_server_resp *resp = &(s_op->resp);
    PVFS_metafile_attr *meta =  &(resp->u.getattr.attr.u.meta);
    int ret = -PVFS_EINVAL;
    job_id_t job_id;
    int i;

   /* Did we find mirror.copies? */
    if (js_p->error_code < 0)
    {
        gossip_debug(GOSSIP_MIRROR_DEBUG,"\tNumber of mirrored copies cannot "
                                         "be retrieved.\n");
        if (resp->u.getattr.attr.mask & (PVFS_ATTR_META_DFILES |
                                         PVFS_ATTR_META_DIST) )
        {
            resp->u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;
            js_p->error_code = SKIP_NEXT_STATE;
        } else {
            resp->u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;
            resp->u.getattr.attr.mask &= ~PVFS_ATTR_META_DFILES;
            resp->u.getattr.attr.mask &= ~PVFS_ATTR_META_DIST;
        }
        memset(&(s_op->key),0,sizeof(s_op->key));
        memset(&(s_op->val),0,sizeof(s_op->val));
        s_op->free_val = 0;

        return SM_ACTION_COMPLETE;
    }

    gossip_debug(GOSSIP_MIRROR_DEBUG,"\tNumber of mirrored copies count "
                                     "successfully retrieved.\n");

   /* check number of mirrored copies */
    if (meta->mirror_copies_count == 0)
    {
       gossip_debug(GOSSIP_MIRROR_DEBUG,"\tNumber of mirrored copies "
                                        "is ZERO.\n");
       gossip_lerr("Mirror handles requested, but number of mirrored copies "
                   "is zero.\n");
       resp->u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;
       js_p->error_code = SKIP_NEXT_STATE;
       return SM_ACTION_COMPLETE;
    }

    gossip_debug(GOSSIP_MIRROR_DEBUG,"\tNumber of mirrored copies "
                                     "retrieved : %d\n"
                                    ,meta->mirror_copies_count);


   /* check to see if total number of mirrored handles is sane */
   if ( (meta->dfile_count * meta->mirror_copies_count) >
         PVFS_REQ_LIMIT_MIRROR_DFILE_COUNT )
   {
       gossip_lerr("Number of mirrored handles(%d) exceeds the system "
                   "limit(%d)\n"
                  ,meta->dfile_count * meta->mirror_copies_count
                  ,PVFS_REQ_LIMIT_MIRROR_DFILE_COUNT); 
       resp->u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;
       js_p->error_code = SKIP_NEXT_STATE;
       return SM_ACTION_COMPLETE;
   }

   js_p->error_code = 0;
   /* get mirrored handles and status of each handle */

   /* initialize */
   if (s_op->free_val)
      free(s_op->val.buffer);
   memset(&(s_op->key),0,sizeof(s_op->key));
   memset(&(s_op->val),0,sizeof(s_op->val));

   for (i=0; i<s_op->keyval_count; i++)
       if (s_op->val_a && s_op->val_a[i].buffer && s_op->free_val)
           free(s_op->val_a[i].buffer);
   if (s_op->val_a)
       free(s_op->val_a);
   if (s_op->key_a)
       free(s_op->key_a);
   if (s_op->error_a)
      free(s_op->error_a);
   s_op->free_val = 0;

   /* allocate space for keys and values */
   s_op->keyval_count = 2;
   s_op->free_val = 1;
   s_op->key_a = s_op->val_a = NULL;
   s_op->error_a = NULL;
   
   s_op->key_a   = malloc(sizeof(*s_op->key_a)   * s_op->keyval_count);
   s_op->val_a   = malloc(sizeof(*s_op->val_a)   * s_op->keyval_count);
   s_op->error_a = malloc(sizeof(*s_op->error_a) * s_op->keyval_count);
   if (!s_op->key_a || !s_op->val_a || !s_op->error_a)
   {
      gossip_lerr("Cannot allocate memory for key/val/error.\n");
      js_p->error_code = -PVFS_ENOMEM;
      goto error_exit;
   }
   memset(s_op->key_a,0,sizeof(*s_op->key_a));
   memset(s_op->val_a,0,sizeof(*s_op->val_a));

   /* set key = user.pvfs2.mirror.handles */
   s_op->key_a[0].buffer    = Trove_Special_Keys[MIRROR_HANDLES_KEY].key;
   s_op->key_a[0].buffer_sz = Trove_Special_Keys[MIRROR_HANDLES_KEY].size;

   /* setup buffer space for handles */
   s_op->val_a[0].buffer = malloc(sizeof(PVFS_handle) * 
                                  meta->dfile_count   *
                                  meta->mirror_copies_count);
   if (!s_op->val_a[0].buffer)
   {
      gossip_lerr("Cannot allocate memory for mirrored handles.\n");
      js_p->error_code = -PVFS_ENOMEM;
      goto error_exit;
   }
   memset(s_op->val_a[0].buffer,0,sizeof(PVFS_handle) *
                                  meta->dfile_count   *
                                  meta->mirror_copies_count);
   s_op->val_a[0].buffer_sz = sizeof(PVFS_handle) *
                              meta->dfile_count   *
                              meta->mirror_copies_count;

   /* set key = user.pvfs2.mirror.status */
   s_op->key_a[1].buffer    = Trove_Special_Keys[MIRROR_STATUS_KEY].key;
   s_op->key_a[1].buffer_sz = Trove_Special_Keys[MIRROR_STATUS_KEY].size;

   /* setup buffer space for handle statuses */
   s_op->val_a[1].buffer = malloc(sizeof(PVFS_error) *
                                  meta->dfile_count  *
                                  meta->mirror_copies_count);
   if (!s_op->val_a[1].buffer)
   {
       gossip_lerr("Cannot allocate memory for mirrored handle statuses.\n");
       js_p->error_code = -PVFS_ENOMEM;
       goto error_exit;
   }
   memset(s_op->val_a[1].buffer,0,sizeof(PVFS_error) *
                                  meta->dfile_count  *
                                  meta->mirror_copies_count);
   s_op->val_a[1].buffer_sz = sizeof(PVFS_error) *
                              meta->dfile_count  *
                              meta->mirror_copies_count;

   
   /* call job to retrieve the key/val pairs */
   ret = job_trove_keyval_read_list(
          s_op->u.getattr.fs_id
         ,s_op->u.getattr.handle
         ,s_op->key_a
         ,s_op->val_a
         ,s_op->error_a
         ,s_op->keyval_count
         ,0
         ,NULL
         ,smcb
         ,0
         ,js_p
         ,&job_id
         ,server_job_context
         ,s_op->req->hints );

   return ret;

error_exit:
   for (i=0; i<s_op->keyval_count; i++)
   {
       if (s_op->val_a && s_op->val_a[i].buffer)
          free(s_op->val_a[i].buffer);
   }
   if (s_op->val_a)
       free(s_op->val_a);
   if (s_op->key_a)
       free(s_op->key_a);
   if (s_op->error_a)
       free(s_op->error_a);
   s_op->val_a = s_op->key_a = NULL;
   s_op->error_a = NULL;
   s_op->keyval_count = 0;
   s_op->free_val = 0;

   resp->u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;

   return SM_ACTION_COMPLETE;
}/*end getattr_read_mirrored_handles_if_required*/




static PINT_sm_action getattr_mirrored_handles_safety_check(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
   gossip_debug(GOSSIP_MIRROR_DEBUG,"Executing %s ...\n",__func__);

    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    struct PVFS_server_resp *resp = &(s_op->resp);
    PVFS_metafile_attr *meta =  &(resp->u.getattr.attr.u.meta);
    int row,col,index;
    int i;


   js_p->error_code = 0;

   /* Check the error code for each key/val pair. */
   for (i=0; i<s_op->keyval_count; i++)
   {
       if (s_op->error_a[i] != 0)
       {
           gossip_debug(GOSSIP_MIRROR_DEBUG,"\tRetrieval of key(%s) failed.\n"
                                           ,(char *)s_op->key_a[i].buffer);
           js_p->error_code = s_op->error_a[i];
       }
   }

   if (js_p->error_code)
   {
      goto error_exit;
   }

   gossip_debug(GOSSIP_MIRROR_DEBUG,"\tWe successfully retrieved handles and "
                                    "statuses.\n");

   for (i=0; i<s_op->keyval_count; i++)
   {  /* Did we get the data that we were expecting from 
       * user.pvfs2.mirror.handles(i=0) or 
       * user.pvfs2.mirror.statuses(i=1)? 
       */

     if (s_op->val_a[i].read_sz != s_op->val_a[i].buffer_sz)
     {
         gossip_lerr("Error: %s key found val size: %d when "
                    "expecting val size: %d\n"
                    ,(char *)s_op->key_a[i].buffer
                    ,s_op->val_a[i].read_sz
                    ,s_op->val_a[i].buffer_sz);
         js_p->error_code = s_op->val_a[i].buffer_sz;
     }
   }/*end for*/

   if (js_p->error_code)
   {
       goto error_exit;
   } 

   /*initialize permanent data structures*/
   meta->mirror_dfile_array = s_op->val_a[0].buffer;
   s_op->u.getattr.mirror_dfile_status_array = s_op->val_a[1].buffer;
   s_op->val_a[0].buffer = s_op->val_a[1].buffer = NULL;

   /* Check the mirroring status for each handle.  If the status is non-zero,
    * the handle is not valid, so put a null in the mirror array for that
    * handle.  Otherwise, do nothing.
   */
   for (row=0; row<meta->mirror_copies_count; row++)
   {
      for (col=0; col<meta->dfile_count; col++)
      {
         index = (row*meta->dfile_count) + col;
         if ( s_op->u.getattr.mirror_dfile_status_array[index] == STATUS_INIT )
              PVFS_handle_clear(meta->mirror_dfile_array[index]);
         gossip_debug(GOSSIP_MIRROR_DEBUG,
                      "\tmirror handle[%d]:%llu \t"
                      "status:%d\n"
                      ,index
                      ,llu(meta->mirror_dfile_array[index])
                      ,s_op->u.getattr.mirror_dfile_status_array[index]);
      }
   }

   /*Cleanup*/
   free(s_op->key_a);
   free(s_op->val_a);
   free(s_op->error_a);
   s_op->key_a = s_op->val_a = NULL;
   s_op->error_a = NULL;
   s_op->keyval_count = 0;
   s_op->free_val = 0;   

   js_p->error_code = 0;
   return SM_ACTION_COMPLETE;

error_exit:
   /* if we have an error, cleanup, and pretend that we never attempted
    * mirrors in the first place.
   */
      gossip_debug(GOSSIP_MIRROR_DEBUG,"\tCleaning up mirror operation...\n");
      js_p->error_code = 0;
      for (i=0; i<s_op->keyval_count; i++)
      {
          if (s_op->val_a[i].buffer)
              free(s_op->val_a[i].buffer);
      }
      free(s_op->key_a);
      free(s_op->val_a);
      free(s_op->error_a);
      s_op->key_a = s_op->val_a = NULL;
      s_op->error_a = NULL;
      s_op->keyval_count = 0; 
      
      /*We MUST set the number of copies to zero to prevent encoding errors
       *later.
      */
      meta->mirror_copies_count = 0;
      meta->mirror_dfile_array = NULL;
      s_op->u.getattr.mirror_dfile_status_array = NULL;
      resp->u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;
      return SM_ACTION_COMPLETE;
}/*end getattr_mirrored_handles_safety_check*/




static PINT_sm_action getattr_read_metafile_distribution_if_required(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -PVFS_EINVAL;
    job_id_t i;

    assert(s_op->attr.objtype == PVFS_TYPE_METAFILE);

    js_p->error_code = 0;

    /* if we don't need to fill in the distribution, skip it */
    if (!(s_op->u.getattr.attrmask & PVFS_ATTR_META_DIST))
    {
        gossip_debug(GOSSIP_GETATTR_DEBUG, "skipping data handle "
                     "distribution read\n");
        return SM_ACTION_COMPLETE;
    }

    s_op->key.buffer = Trove_Common_Keys[METAFILE_DIST_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[METAFILE_DIST_KEY].size;

    /*
      there *should* be some distribution information.  if not, dump
      which handle is busted and assertion die for now while we're not
      handling this kind of error
    */
    if (s_op->resp.u.getattr.attr.u.meta.dist_size < 1)
    {
        gossip_err("Cannot Read Dist!  Got an invalid dist size for "
                   "handle %llu,%d\n",llu(s_op->u.getattr.handle),
                   s_op->u.getattr.fs_id);
        js_p->error_code = -PVFS_EINVAL;
        return SM_ACTION_COMPLETE;
    }
    assert(s_op->resp.u.getattr.attr.u.meta.dist_size > 0);

    /* add mask value to indicate the distribution is filled */
    s_op->resp.u.getattr.attr.mask |= PVFS_ATTR_META_DIST;

    if(s_op->free_val)
    {
        free(s_op->val.buffer);
    }
    s_op->val.buffer_sz = s_op->resp.u.getattr.attr.u.meta.dist_size; 
    s_op->val.buffer = malloc(s_op->val.buffer_sz);
    if (!s_op->val.buffer)
    {
        gossip_err("Cannot allocate dist of size %d\n",
                   s_op->val.buffer_sz);
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    s_op->free_val = 1;

    ret = job_trove_keyval_read(
        s_op->u.getattr.fs_id, s_op->u.getattr.handle,
        &(s_op->key), &(s_op->val),
        0,
        NULL,
        smcb, 0, js_p, &i, server_job_context, s_op->req->hints);

    return ret;
}

static PINT_sm_action getattr_read_stuffed_size(
    struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    job_id_t job_id;

    if(js_p->error_code == -TROVE_ENOENT)
    {
        gossip_debug(
            GOSSIP_GETATTR_DEBUG, "Getattr detected non-stuffed file.\n");
        /* this means that the keyval fields used to indicate a file is
         * stuffed are not present.  Set mask accordingly and continue.
         */
        s_op->resp.u.getattr.attr.mask |= PVFS_ATTR_META_UNSTUFFED;
        js_p->error_code = 0;
        return SM_ACTION_COMPLETE;
    }
    if(js_p->error_code)
    {
        /* any other error code here is just a normal error case */
        /* preserve error code and catch next error transition */
        return SM_ACTION_COMPLETE;
    }

    gossip_debug(
        GOSSIP_GETATTR_DEBUG, "Getattr detected stuffed file.\n");
    /* otherwise, we found keyval fields indicating that the file is
     * stuffed.  It does not matter if the client asked for the size or not;
     * we must retrieve a valid stuffed_size value for the attrs.
     */
    s_op->resp.u.getattr.attr.mask &= (~(PVFS_ATTR_META_UNSTUFFED));

    return(job_trove_dspace_getattr(
        s_op->u.getattr.fs_id,
        s_op->resp.u.getattr.attr.u.meta.dfile_array[0],
        smcb,
        &s_op->ds_attr,
        0,
        js_p,
        &job_id,
        server_job_context,
        s_op->req->hints));
}

static PINT_sm_action getattr_interpret_stuffed_size(
    struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_metafile_attr *meta    = &(s_op->resp.u.getattr.attr.u.meta);

    if(js_p->error_code == 0)
    {
        meta->stuffed_size = s_op->ds_attr.u.datafile.b_size;
    }

    /* deliberately leave error_code unchanged so that any errors get 
     * handled in the next state
     */
    return SM_ACTION_COMPLETE;
}


/* interpret_metafile_distribution()
 *
 * capture and encode results of reading distribution
 */
static PINT_sm_action interpret_metafile_distribution(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_object_attr *resp_attr = &s_op->resp.u.getattr.attr;
    
    if(js_p->error_code < 0)
    {
        return SM_ACTION_COMPLETE;
    }

    if(s_op->u.getattr.attrmask & PVFS_ATTR_META_DIST)
    {
        /* successfully read dist key; make sure we got something valid */
        if(s_op->val.read_sz != s_op->val.buffer_sz)
        {
            gossip_err("Error: %s key found val size: %d when "
                       "expecting val size: %d\n",
                Trove_Common_Keys[METAFILE_DIST_KEY].key,
                s_op->val.read_sz,
                s_op->val.buffer_sz);

            /* clear bitmask to prevent double free between setup_resp and
             * PINT_free_object_attr()
             */
            s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_DIST;

            js_p->error_code = -PVFS_EIO;
            return SM_ACTION_COMPLETE;
        }

        assert(s_op->val.buffer);
        PINT_dist_decode(&resp_attr->u.meta.dist, s_op->val.buffer);

        if(resp_attr->u.meta.dist == 0) {
            gossip_err("Found dist of 0 for handle %llu,%d\n",
                    llu(s_op->u.getattr.handle), s_op->u.getattr.fs_id);
            PVFS_perror("Metafile getattr_setup_resp",js_p->error_code);
            js_p->error_code = -PVFS_EIO;
            return SM_ACTION_COMPLETE;
        }
    }

    js_p->error_code = 0;
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_setup_resp(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_object_attr *resp_attr = &s_op->resp.u.getattr.attr;

    if(js_p->error_code > 0)
    {
        /* if we reach this state with a positive error code it means that
         * nothing is wrong; we just used one of the explicit STATE_*
         * transitions
         */
        js_p->error_code = 0;
    }
    if(js_p->error_code < 0)
    {
        free_nested_getattr_data(s_op);
        return SM_ACTION_COMPLETE;
    }

    gossip_debug(
        GOSSIP_GETATTR_DEBUG,
        "-  retrieved attrs: [owner = %d, group = %d\n\t"
        "perms = %o, type = %d, atime = %llu, mtime = %llu\n\t"
        "ctime = %llu, dist_size = %d]\n",
        resp_attr->owner, resp_attr->group, resp_attr->perms,
        resp_attr->objtype, llu(resp_attr->atime),
        llu(resp_attr->mtime), llu(resp_attr->ctime),
        (int)resp_attr->u.meta.dist_size);

    if (resp_attr->objtype == PVFS_TYPE_METAFILE)
    {
        if (resp_attr->mask & PVFS_ATTR_META_DFILES)
        {
            if (resp_attr->u.meta.dfile_count)
            {
                assert(resp_attr->u.meta.dfile_array);
            }
            gossip_debug(GOSSIP_GETATTR_DEBUG,
                         "  also returning %d datafile handles\n",
                         resp_attr->u.meta.dfile_count);
        }
        if (resp_attr->mask & PVFS_ATTR_META_MIRROR_DFILES)
        {
           if (resp_attr->u.meta.mirror_copies_count)
              assert(resp_attr->u.meta.mirror_dfile_array);
           gossip_debug(GOSSIP_GETATTR_DEBUG,
                        "  also returning %d mirrored copies\n"
                       ,resp_attr->u.meta.mirror_copies_count);
        }
        if (resp_attr->mask & PVFS_ATTR_META_DIST)
        {
            /* we have already gathered the dist field in an earlier state */
            gossip_debug(GOSSIP_GETATTR_DEBUG,
                         "  also returning dist size of %d\n",
                         resp_attr->u.meta.dist_size);
        }
    }
    else if ((resp_attr->objtype == PVFS_TYPE_DATAFILE) &&
             (resp_attr->mask & PVFS_ATTR_DATA_SIZE))
    {
        gossip_debug(GOSSIP_GETATTR_DEBUG,
                     "  also returning data size of %lld\n",
                     lld(resp_attr->u.data.size));
    }
    else if ((resp_attr->objtype == PVFS_TYPE_SYMLINK) &&
             (resp_attr->mask & PVFS_ATTR_SYMLNK_TARGET))
    {
        if (js_p->error_code == 0)
        {
            assert(resp_attr->u.sym.target_path);
            assert(resp_attr->u.sym.target_path_len);
            /*
              adjust target path len down to actual size ; always
              include the null termination char in the target_path_len
            */
            resp_attr->u.sym.target_path_len =
                (strlen(resp_attr->u.sym.target_path) + 1);

            gossip_debug(GOSSIP_GETATTR_DEBUG,
                         "  also returning link target of %s (len %d)\n",
                         resp_attr->u.sym.target_path,
                         resp_attr->u.sym.target_path_len);
        }
        else
        {
            gossip_err("Failed to retrieve symlink target path for "
                       "handle %llu,%d\n",llu(s_op->u.getattr.handle),
                       s_op->u.getattr.fs_id);
            PVFS_perror("Symlink retrieval failure",js_p->error_code);

            free_nested_getattr_data(s_op);
            js_p->error_code = -PVFS_EINVAL;
            return SM_ACTION_COMPLETE;
        }
    }
    else if ((resp_attr->objtype == PVFS_TYPE_DIRECTORY) &&
            (resp_attr->mask & PVFS_ATTR_DIR_HINT))
    {
        gossip_debug(GOSSIP_GETATTR_DEBUG, " server returning "
            "dirent_count = %llu "
            "dfile_count = %d "
            "dist_name_len    = %d "
            "dist_params_len  = %d\n",
            llu(resp_attr->u.dir.dirent_count),
            resp_attr->u.dir.hint.dfile_count,
            resp_attr->u.dir.hint.dist_name_len,
            resp_attr->u.dir.hint.dist_params_len);
    }

    gossip_debug(GOSSIP_GETATTR_DEBUG,"@ End %s attributes: sending "
                 "status %d (error = %d)\n",
                 PINT_util_get_object_type(resp_attr->objtype),
                 s_op->resp.status, js_p->error_code);

#if 0
    gossip_debug(GOSSIP_GETATTR_DEBUG, "returning attrmask ");
    PINT_attrmask_print(GOSSIP_GETATTR_DEBUG,
                        s_op->resp.u.getattr.attr.mask);
#endif

    free_nested_getattr_data(s_op);
    return SM_ACTION_COMPLETE;
}

static void free_nested_getattr_data(struct PINT_server_op *s_op)
{
   int i;
    /* free up anything that was set up specifically by this nested machine */
    if (s_op->free_val)
    {
       for (i=0; i<s_op->keyval_count; i++)
       {
           if (s_op->val_a[i].buffer)
               free(s_op->val_a[i].buffer);
       }
    }
    if(s_op->val_a)
    {
        free(s_op->val_a);
        s_op->val_a = NULL;
    }
    if(s_op->key_a)
    {
        free(s_op->key_a);
        s_op->key_a = NULL;
    }
    if(s_op->u.getattr.err_array)
    {
        free(s_op->u.getattr.err_array);
        s_op->u.getattr.err_array = NULL;
    }
    if (s_op->u.getattr.mirror_dfile_status_array)
    {
        free(s_op->u.getattr.mirror_dfile_status_array);
        s_op->u.getattr.mirror_dfile_status_array = NULL;
    }
    if(s_op->free_val)
    {
        free(s_op->val.buffer);
        s_op->val.buffer = NULL;
    }
    if(s_op->error_a)
    {
        free(s_op->error_a);
        s_op->error_a = NULL;
    }

    return;
}

static PINT_sm_action getattr_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);

    PINT_free_object_attr(&s_op->resp.u.getattr.attr);
    return(server_state_machine_complete(smcb));
}

static PINT_sm_action getattr_with_prelude_init(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    js_p->error_code = 0;
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_setup_op(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PVFS_handle_copy(s_op->u.getattr.handle, s_op->req->u.getattr.handle);
    s_op->u.getattr.fs_id = s_op->req->u.getattr.fs_id;
    s_op->u.getattr.attrmask = s_op->req->u.getattr.attrmask;
    s_op->u.getattr.err_array = NULL;
    s_op->u.getattr.mirror_dfile_status_array = NULL;

    js_p->error_code = 0;
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_datafile_handles_safety_check(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);

    if((js_p->error_code == 0) &&
        (s_op->u.getattr.attrmask & PVFS_ATTR_META_DFILES))
    {
        /* successfully read datafile key; make sure we got something valid */
        if(s_op->val.read_sz != s_op->val.buffer_sz)
        {
            gossip_err("Error: %s key found val size: %d when "
                       "expecting val size: %d\n",
                Trove_Common_Keys[METAFILE_HANDLES_KEY].key,
                s_op->val.read_sz,
                s_op->val.buffer_sz);

            /* clear bitmask to prevent double free between setup_resp and
             * PINT_free_object_attr()
             */
            s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_DFILES;
            s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_DIST;
            s_op->resp.u.getattr.attr.mask &= ~PVFS_ATTR_META_MIRROR_DFILES;

            js_p->error_code = -PVFS_EIO;
            return SM_ACTION_COMPLETE;
        }
    }

    /* otherwise deliberately preserve existing error code */
    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_get_dirdata_handle(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret;
    job_id_t tmp_id;

    s_op->key.buffer = Trove_Common_Keys[DIR_ENT_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[DIR_ENT_KEY].size;
    if(s_op->free_val)
    {
        free(s_op->val.buffer);
    }
    s_op->val.buffer = &s_op->u.getattr.dirent_handle;
    s_op->val.buffer_sz = sizeof(PVFS_handle);
    s_op->free_val = 0;

    ret = job_trove_keyval_read(
        s_op->u.getattr.fs_id, s_op->u.getattr.handle,
        &s_op->key, &s_op->val,
        0,
        NULL,
        smcb,
        0,
        js_p,
        &tmp_id,
        server_job_context, s_op->req->hints);

    return ret;
}
        
static PINT_sm_action getattr_get_dirent_count(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret;
    job_id_t tmp_id;

    if (!(s_op->u.getattr.attrmask & PVFS_ATTR_DIR_DIRENT_COUNT))
    {
         /* the caller didn't really want the dirent count; skip to get
          * directory hints
          */
         js_p->error_code = STATE_DIR_HINT;
         return SM_ACTION_COMPLETE;
    }
    ret = job_trove_keyval_get_handle_info(
        s_op->u.getattr.fs_id,
        s_op->u.getattr.dirent_handle,
        TROVE_KEYVAL_HANDLE_COUNT |
        0,
        &s_op->u.getattr.keyval_handle_info,
        smcb,
        0,
        js_p,
        &tmp_id,
        server_job_context, s_op->req->hints);

    return ret;
}

static PINT_sm_action getattr_interpret_dirent_count(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    switch(js_p->error_code)
    {
        case -TROVE_ENOENT:
            js_p->error_code = 0;
            s_op->resp.u.getattr.attr.u.dir.dirent_count = 0;
            break;
        case 0:
            s_op->resp.u.getattr.attr.u.dir.dirent_count =
                s_op->u.getattr.keyval_handle_info.count;
            break;
        default:
            return SM_ACTION_COMPLETE;
    }

    gossip_debug(GOSSIP_GETATTR_DEBUG, "getattr: dirent_count: %lld\n",
        lld(s_op->resp.u.getattr.attr.u.dir.dirent_count));

    return SM_ACTION_COMPLETE;
}

static PINT_sm_action getattr_get_dir_hint(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret, i;
    job_id_t tmp_id;

    /* NOTE: memory allocations are released in the getattr_cleanup()
     * function 
     */
   
    if (!(s_op->u.getattr.attrmask & PVFS_ATTR_DIR_HINT))
    {
        /* the caller didn't really want the dir hints; skip
         */
        js_p->error_code = STATE_DONE;
        return SM_ACTION_COMPLETE;
    }


    gossip_debug(GOSSIP_SERVER_DEBUG, "  trying to getxattr of %s,%s,%s "
                 "of dir handle (coll_id = %d, handle = %llu\n",
                 Trove_Special_Keys[DIST_NAME_KEY].key,
                 Trove_Special_Keys[DIST_PARAMS_KEY].key, 
                 Trove_Special_Keys[NUM_DFILES_KEY].key,
                 s_op->u.getattr.fs_id, llu(s_op->u.getattr.handle));

    s_op->resp.u.getattr.attr.u.dir.hint.dist_params = 
        (char *) calloc(1, PVFS_REQ_LIMIT_DIST_BYTES);
    if (!s_op->resp.u.getattr.attr.u.dir.hint.dist_params)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    s_op->resp.u.getattr.attr.u.dir.hint.dist_params_len = 
        PVFS_REQ_LIMIT_DIST_BYTES;

    s_op->resp.u.getattr.attr.u.dir.hint.dist_name = 
        (char *) calloc(1, PVFS_REQ_LIMIT_DIST_NAME);
    if (!s_op->resp.u.getattr.attr.u.dir.hint.dist_name)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    s_op->resp.u.getattr.attr.u.dir.hint.dist_name_len   = 
        PVFS_REQ_LIMIT_DIST_NAME;

    s_op->key_a = 
        (PVFS_ds_keyval *) calloc(NUM_SPECIAL_KEYS, sizeof(PVFS_ds_keyval));
    if (s_op->key_a == NULL)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    s_op->val_a = (PVFS_ds_keyval *) calloc(NUM_SPECIAL_KEYS
                                           ,sizeof(PVFS_ds_keyval));
    if (s_op->val_a == NULL)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    s_op->u.getattr.err_array = (PVFS_error*)calloc(NUM_SPECIAL_KEYS,
        sizeof(PVFS_error));
    if(s_op->u.getattr.err_array == NULL)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;

    }

    s_op->free_val = 0;
    s_op->keyval_count = NUM_SPECIAL_KEYS;
    for (i = 0; i < NUM_SPECIAL_KEYS; i++)
    {
        s_op->key_a[i].buffer = Trove_Special_Keys[i].key;
        s_op->key_a[i].buffer_sz = Trove_Special_Keys[i].size;
        if (i == NUM_DFILES_KEY)
        {
            s_op->val_a[i].buffer = (char *) calloc(1, 16);
            if(s_op->val_a[i].buffer == NULL)
            {
                js_p->error_code = -PVFS_ENOMEM;
                return SM_ACTION_COMPLETE;
            }
            s_op->val_a[i].buffer_sz = 16;
        }
        else if (i == DIST_PARAMS_KEY) {
            s_op->val_a[i].buffer 
            = s_op->resp.u.getattr.attr.u.dir.hint.dist_params;
            s_op->val_a[i].buffer_sz 
            = s_op->resp.u.getattr.attr.u.dir.hint.dist_params_len;
        }
        else if (i == DIST_NAME_KEY) {
            s_op->val_a[i].buffer 
            = s_op->resp.u.getattr.attr.u.dir.hint.dist_name;
            s_op->val_a[i].buffer_sz 
            = s_op->resp.u.getattr.attr.u.dir.hint.dist_name_len;
        }
    }

    js_p->error_code = 0;
    ret = job_trove_keyval_read_list(
        s_op->u.getattr.fs_id, 
        s_op->u.getattr.handle,
        s_op->key_a, s_op->val_a, s_op->u.getattr.err_array, NUM_SPECIAL_KEYS,
        0, NULL, smcb, 0, js_p, &tmp_id,
        server_job_context, s_op->req->hints);

    return ret;
}

static PINT_sm_action getattr_interpret_dir_hint(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    if(js_p->error_code != 0 && js_p->error_code != -TROVE_ENOENT)
    {
        /* if we failed to get any of the keys, and the error code is due to
         * something other than the keys simply not being present, then
         * propigate the error.
         */
        return SM_ACTION_COMPLETE;
    }

    gossip_debug(GOSSIP_SERVER_DEBUG, 
        "getattr: job status code = %d\n", js_p->error_code);
    if (s_op->val_a && s_op->key_a)
    {
        long int dfile_count = 0;

        if (s_op->u.getattr.err_array[DIST_NAME_KEY] == 0)
        {
            gossip_debug(GOSSIP_SERVER_DEBUG, 
                "val_a[DIST_NAME_KEY] %p read_sz = %d dist_name = %s\n", 
                s_op->val_a[DIST_NAME_KEY].buffer, 
                s_op->val_a[DIST_NAME_KEY].read_sz,
                (char *)s_op->val_a[DIST_NAME_KEY].buffer);
            s_op->resp.u.getattr.attr.u.dir.hint.dist_name_len = 
                s_op->val_a[DIST_NAME_KEY].read_sz;
        }
        else
        {
            s_op->resp.u.getattr.attr.u.dir.hint.dist_name_len = 0;
        }            
        s_op->val_a[DIST_NAME_KEY].buffer = NULL;

        if (s_op->u.getattr.err_array[DIST_PARAMS_KEY] == 0)
        {
            gossip_debug(GOSSIP_SERVER_DEBUG, 
                "val_a[DIST_PARAMS_KEY] %p read_sz = %d dist_params = %s\n", 
                s_op->val_a[DIST_PARAMS_KEY].buffer, 
                s_op->val_a[DIST_PARAMS_KEY].read_sz,
                (char *)s_op->val_a[DIST_PARAMS_KEY].buffer);
            s_op->resp.u.getattr.attr.u.dir.hint.dist_params_len = 
                s_op->val_a[DIST_PARAMS_KEY].read_sz;
        }
        else
        {
            s_op->resp.u.getattr.attr.u.dir.hint.dist_params_len = 0;
        }
        if (s_op->val_a[DIST_PARAMS_KEY].buffer)
        {
           s_op->val_a[DIST_PARAMS_KEY].buffer = NULL;
        }



        if (s_op->u.getattr.err_array[NUM_DFILES_KEY] == 0)
        {
            char *endptr = NULL;
            gossip_debug(GOSSIP_SERVER_DEBUG, "val_a[NUM_DFILES_KEY] %p "
                                              "read_sz = %d\n", 
                s_op->val_a[NUM_DFILES_KEY].buffer, 
                s_op->val_a[NUM_DFILES_KEY].read_sz);
            dfile_count = strtol(s_op->val_a[NUM_DFILES_KEY].buffer
                                , &endptr, 10);
            if (*endptr != '\0' || dfile_count < 0)
            {
                dfile_count = 0;
            }
        }
        if(s_op->val_a[NUM_DFILES_KEY].buffer)
        {
            free(s_op->val_a[NUM_DFILES_KEY].buffer);
            s_op->val_a[NUM_DFILES_KEY].buffer = NULL;
            s_op->val_a[NUM_DFILES_KEY].buffer_sz = 0;
        }
        s_op->keyval_count = 0;
        s_op->free_val = 0;


        s_op->resp.u.getattr.attr.u.dir.hint.dfile_count = dfile_count;

        gossip_debug(GOSSIP_SERVER_DEBUG, "getattr: dir hint dfile_count: %d\n",
            s_op->resp.u.getattr.attr.u.dir.hint.dfile_count);

        js_p->error_code = 0;
    }/* end if val_a and key_a */
    return SM_ACTION_COMPLETE;
}

/* getattr_detect_stuffed()
 *
 * determine if a file is stuffed or not
 */
static PINT_sm_action getattr_detect_stuffed(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    job_id_t tmp_id;

    /* we can determine stuffedness by the presence of the dfiles req key */

    s_op->key.buffer = Trove_Common_Keys[NUM_DFILES_REQ_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[NUM_DFILES_REQ_KEY].size;
    if(s_op->free_val)
    {
        free(s_op->val.buffer);
    }
    s_op->val.buffer =  &s_op->u.getattr.num_dfiles_req;
    s_op->val.buffer_sz =  sizeof(s_op->u.getattr.num_dfiles_req);
    s_op->free_val = 0;

    return(job_trove_keyval_read(
        s_op->u.getattr.fs_id, 
        s_op->u.getattr.handle,
        &(s_op->key), 
        &(s_op->val), 
        0, 
        NULL, smcb, 0, js_p,
        &tmp_id, server_job_context,
        s_op->req->hints));
}

PINT_GET_OBJECT_REF_DEFINE(getattr);

struct PINT_server_req_params pvfs2_get_attr_params =
{
    .string_name = "getattr",
    .perm = PINT_SERVER_CHECK_ATTR,
    .access_type = PINT_server_req_readonly,
    .sched_policy = PINT_SERVER_REQ_SCHEDULE,
    .get_object_ref = PINT_get_object_ref_getattr,
    .state_machine = &pvfs2_get_attr_sm
};

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