root/branches/Orange-Elaine-Distr-Dir-Branch/src/server/lookup.sm @ 8492

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

#include <string.h>
#include <stddef.h>
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#ifdef HAVE_MALLOC_H
#include <malloc.h>
#endif

#include "server-config.h"
#include "pvfs2-server.h"
#include "pvfs2-attr.h"
#include "str-utils.h"
#include "pint-util.h"
#include "pvfs2-internal.h"
#include "check.h"

enum 
{
    STATE_ENOTDIR = 22,
    STATE_NOMORESEGS = 23,
    LOOKUP_CHECK_DIR_ACLS = 24,
};

%%

machine pvfs2_lookup_sm
{
    state prelude
    {
        jump pvfs2_prelude_sm;
        success => init;
        default => final_response;
    }

    state init
    {
        run lookup_init;
        STATE_ENOTDIR => setup_resp;
        default => read_object_metadata;
    }

    state read_object_metadata
    {
        run lookup_read_object_metadata;
        success => verify_object_metadata;
        default => setup_resp;
    }

    state verify_object_metadata
    {
        run lookup_verify_object_metadata;
        LOOKUP_CHECK_DIR_ACLS => read_directory_acls;
        success => read_num_directory_entry_handles;
        default => setup_resp;
    }

    state read_directory_acls
    {
        run lookup_check_acls_if_needed;
        default => check_acls;
    }

    state check_acls
    {
        run lookup_check_acls;
        success => read_num_directory_entry_handles;
        default => setup_resp;
    }

    state read_num_directory_entry_handles
    {
        run lookup_read_num_directory_entry_handles;
        default => read_directory_entry_handles;
    }

    state read_directory_entry_handles
    {
        run lookup_read_directory_entry_handles;
        success => read_directory_entry;
        default => setup_resp;
    }

    state read_directory_entry
    {
        run lookup_read_directory_entry;
        success => read_object_metadata;
        default => setup_resp;
    }
    
    state setup_resp 
    {
        run lookup_setup_resp;
        default => final_response;
    }

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

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

%%

/*
 * Function: lookup_init
 *
 * Synopsis: initializes internal structures and posts job to request
 * scheduler.
 *
 * Assumes req structure holds a valid path.
 *
 * Initializes segp, seg_ct, seg_nr fields in s_op->u.lookup.
 *
 * Allocates memory for handle and attribute arrays that will be
 * returned in the response.
 *
 * Note: memory is allocated as one big chunk, pointed to by
 * s_op->resp.u.lookup_path.handle_array.
 *
 */
static PINT_sm_action lookup_init(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    char *ptr = NULL;


    /* fill in the lookup portion of the PINT_server_op */
    s_op->u.lookup.segp = NULL;
    s_op->u.lookup.seg_nr = 0;
    s_op->u.lookup.seg_ct = PINT_string_count_segments(
        s_op->req->u.lookup_path.path);
    s_op->u.lookup.handle_ct = 0;
    s_op->u.lookup.attr_ct = 0;

    gossip_debug(GOSSIP_SERVER_DEBUG, " STARTING LOOKUP REQUEST "
                     "(path:%s)(fs_id:%d)(handle:%llu)(attrmask:%u)"
                     "(# of segments:%u)\n",
                     s_op->req->u.lookup_path.path,
                     s_op->req->u.lookup_path.fs_id,
                     llu(s_op->req->u.lookup_path.handle),
                     s_op->req->u.lookup_path.attrmask,
                     s_op->u.lookup.seg_ct);
    if ((s_op->u.lookup.seg_ct < 0) ||
        (s_op->u.lookup.seg_ct > PVFS_REQ_LIMIT_PATH_SEGMENT_COUNT))
    {
        gossip_err("  invalid path %s (bad segment count); "
                   "sending error response\n",
                   s_op->req->u.lookup_path.path);
        js_p->error_code = -PVFS_ENOTDIR;
        return SM_ACTION_DEFERRED;
    }

    /* allocate the internal ds_attr_array */
    s_op->u.lookup.ds_attr_array = (PVFS_ds_attributes *)
        malloc(s_op->u.lookup.seg_ct * sizeof(PVFS_ds_attributes));
    if(!s_op->u.lookup.ds_attr_array)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return 1;
    }

    /* allocate memory
     *
     * Note: all memory is allocated in a single block,
     * pointed to by s_op->resp.u.lookup_path.handle_array
     */
    ptr = malloc(s_op->u.lookup.seg_ct *
                 (sizeof(PVFS_handle) + sizeof(PVFS_object_attr)));
    if (!ptr)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }

    s_op->resp.u.lookup_path.handle_array = (PVFS_handle *)ptr;
    ptr += (s_op->u.lookup.seg_ct * sizeof(PVFS_handle));

    s_op->resp.u.lookup_path.attr_array = (PVFS_object_attr *)ptr;

    js_p->error_code = 0;
    return SM_ACTION_COMPLETE;
}

/*
 * Function: lookup_read_object_metadata
 *
 * Synopsis: Given an object handle, looks up the attributes
 * (metadata) for that handle.
 *
 * Initializes key and value structures to direct metadata:
 * - if this is the starting (base) handle, store in
 *   s_op->u.lookup.base_attr
 * - otherwise store it in the appropriate slot in the resp handle array
 *
 * Posts the keyval read to trove.
 */
static PINT_sm_action lookup_read_object_metadata(
        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 j_id;
    PVFS_handle handle = PVFS_HANDLE_NULL;
    PVFS_ds_attributes *ds_attr = NULL;

    assert(s_op->u.lookup.seg_nr <= s_op->u.lookup.seg_ct);

    /* use the base handle if we haven't looked up a segment yet */
    if (s_op->u.lookup.seg_nr == 0)
    {
        handle = s_op->req->u.lookup_path.handle;
        ds_attr = &(s_op->u.lookup.ds_attr_array[0]);
    }
    else
    {
        handle = s_op->resp.u.lookup_path.handle_array[
            s_op->u.lookup.seg_nr-1];
        ds_attr = &(s_op->u.lookup.ds_attr_array[
                        s_op->u.lookup.seg_nr - 1]);
    }

    /* update our successful handle read count */
    s_op->u.lookup.handle_ct++;

    /* Copy the fsid and handle to the s_op structure for the acl check */
    s_op->target_handle = handle;
    s_op->target_fs_id = s_op->req->u.lookup_path.fs_id;

    /* get the dspace attributes/metadata */
    ret = job_trove_dspace_getattr(
        s_op->req->u.lookup_path.fs_id, handle, smcb, ds_attr,
        0, js_p, &j_id, server_job_context, s_op->req->hints );

    return ret;
}

/*
 * Function: lookup_verify_object_metadata
 *
 * Synopsis: Examine the metadata returned from the prelude sm.  If
 * the metadata is for a directory, prepare to read the handle of the
 * next segment, if there is one.  If the metadata is for a file,
 * prepare to send a response.
 *
 * If the object is a directory, this function sets the
 * s_op->u.lookup.segp value to point to the next segment to look up;
 * this is used in lookup_read_directory_entry.
 *
 * This function does not post an operation, but rather returns 1
 * immediately.
 */
static PINT_sm_action lookup_verify_object_metadata(
        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;
    PVFS_object_attr *a_p = NULL;
    PVFS_ds_attributes *ds_attr = NULL;

    if (s_op->u.lookup.seg_nr == 0)
    {
        a_p = &s_op->attr;
        ds_attr = &(s_op->u.lookup.ds_attr_array[0]);
    }
    else
    {
        a_p = &s_op->resp.u.lookup_path.attr_array[
            s_op->u.lookup.seg_nr - 1];
        ds_attr = &(s_op->u.lookup.ds_attr_array[
                        s_op->u.lookup.seg_nr - 1]);
    }

    PVFS_ds_attr_to_object_attr(ds_attr, a_p);
    a_p->mask = PVFS_ATTR_COMMON_ALL;
    s_op->target_object_attr = a_p;

    /* update our successful attr read count */
    s_op->u.lookup.attr_ct++;

    assert(((a_p->objtype == PVFS_TYPE_DIRECTORY) ||
            (a_p->objtype == PVFS_TYPE_METAFILE)  ||
            (a_p->objtype == PVFS_TYPE_SYMLINK)));

    gossip_debug(
        GOSSIP_SERVER_DEBUG, "  attrs = (owner = %d, group = %d, "
        "perms = %o, type = %d)\n", a_p->owner, a_p->group,
        a_p->perms, a_p->objtype);

    /* if we hit a metafile, we are done */
    if (a_p->objtype == PVFS_TYPE_METAFILE)
    {
        gossip_debug(GOSSIP_SERVER_DEBUG, "  object is a metafile; "
                     "halting lookup and sending response\n");

        js_p->error_code = STATE_ENOTDIR;
        return SM_ACTION_COMPLETE;
    }

    /*
      if we hit a symlink, we're done; client will pick up the pieces
      and continue to resolve the symlink if required
    */
    if (a_p->objtype == PVFS_TYPE_SYMLINK)
    {
        gossip_debug(GOSSIP_SERVER_DEBUG, "  object is a symlink; "
                     "halting lookup and sending response\n");

        js_p->error_code = STATE_ENOTDIR;
        return SM_ACTION_COMPLETE;
    }

    /* if we looked up all the segments, we are done */
    if (s_op->u.lookup.seg_nr == s_op->u.lookup.seg_ct)
    {
        gossip_debug(GOSSIP_SERVER_DEBUG, "  no more segments in path; "
                     "sending response\n");

        js_p->error_code = STATE_NOMORESEGS;
        return SM_ACTION_COMPLETE;
    }

    /* if we reach this point, the object is a directory.  Verify that we
     * have execute permission on the directory before continuing traversal
     */
    js_p->error_code = PINT_check_mode(a_p, s_op->req->credentials.uid,
        s_op->req->credentials.gid, PINT_ACCESS_EXECUTABLE);
    if(js_p->error_code != 0)
    {
        /* doesn't look like we have permission to traverse directory; bail
         * out
         */
        js_p->error_code = LOOKUP_CHECK_DIR_ACLS;
        return SM_ACTION_COMPLETE;
    }

    /* find the segment that we should look up in the directory */
    ret = PINT_string_next_segment(
        s_op->req->u.lookup_path.path, &s_op->u.lookup.segp,
        &s_op->u.lookup.segstate);

    if(ret != 0)
    {
        gossip_err("PINT_string_next_segment failed: path: %s\n",
                   s_op->req->u.lookup_path.path);
    }
    assert(ret == 0);

    gossip_debug(GOSSIP_SERVER_DEBUG, "  object is a directory; will be "
                 "looking for handle for segment \"%s\" in a bit\n",
                 s_op->u.lookup.segp);


    return SM_ACTION_COMPLETE;
}

/*
 * Post a keyval DB read of the posix acls to check and see if 
 * directory traversal is allowed or not
 */
static PINT_sm_action lookup_check_acls_if_needed(
    struct PINT_smcb *smcb, job_status_s* js_p)
{
    int ret = -PVFS_EINVAL;
    job_id_t i;
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);

    /* If we get here with an invalid fsid and handle, we have to
     * return -PVFS_EACCESS 
     */
    if (s_op->target_fs_id == PVFS_FS_ID_NULL
        || s_op->target_handle == PVFS_HANDLE_NULL)
    {
        js_p->error_code = -PVFS_EACCES;
        return SM_ACTION_COMPLETE;
    }
    js_p->error_code = 0;

    memset(&s_op->key, 0, sizeof(PVFS_ds_keyval));
    memset(&s_op->val, 0, sizeof(PVFS_ds_keyval));
    s_op->key.buffer = "system.posix_acl_access";
    s_op->key.buffer_sz = strlen(s_op->key.buffer) + 1;
    s_op->val.buffer = (char *) malloc(PVFS_REQ_LIMIT_VAL_LEN);
    if (!s_op->val.buffer)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    s_op->val.buffer_sz = PVFS_REQ_LIMIT_VAL_LEN;

    gossip_debug(GOSSIP_PERMISSIONS_DEBUG, "About to retrieve acl keyvals "
                 "for handle %llu\n", llu(s_op->target_handle));

    /* Read acl keys */
    ret = job_trove_keyval_read(
        s_op->target_fs_id,
        s_op->target_handle,
        &s_op->key,
        &s_op->val,
        0,
        NULL,
        s_op,
        0,
        js_p,
        &i,
        server_job_context, s_op->req->hints);
    return ret;
}

/*
 * Verify if the completed keyval DB operation allows the lookup
 * to proceed or not. i.e. executable privileges on directory
 * for the requesting user or not.
 */
static PINT_sm_action lookup_check_acls(
    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 *obj_attr = NULL;
    int want = PVFS2_ACL_EXECUTE;

    /* The dspace attr must have been read at this point */
    obj_attr = s_op->target_object_attr;
    assert(obj_attr);

    /* anything non-zero we treat as a real error */
    if (js_p->error_code)
    {
        goto cleanup;
    }
    /* ok; let the actual acl check be done */
    js_p->error_code = PINT_check_acls(s_op->val.buffer,
                        s_op->val.read_sz,
                        obj_attr, 
                        s_op->req->credentials.uid,
                        s_op->req->credentials.gid,
                        want);
    /* if we are good to go,
       find the segment that we should look up in the directory */
    if (js_p->error_code == 0)
    {
        js_p->error_code = PINT_string_next_segment(
            s_op->req->u.lookup_path.path, &s_op->u.lookup.segp,
            &s_op->u.lookup.segstate);

        if(js_p->error_code != 0)
        {
            gossip_err("PINT_string_next_segment failed to get the"
                       "next segment to lookup from the path: %s\n",
                       s_op->req->u.lookup_path.path);
        }
        else
        {
            gossip_debug(GOSSIP_SERVER_DEBUG, "  after ACL check "
                         "object is a directory; will be "
                         "looking for handle for segment \"%s\" in a bit\n",
                         s_op->u.lookup.segp);
        }
    }
cleanup:
    if (s_op->val.buffer) 
        free(s_op->val.buffer);
    memset(&s_op->key, 0, sizeof(PVFS_ds_keyval));
    memset(&s_op->val, 0, sizeof(PVFS_ds_keyval));
    return SM_ACTION_COMPLETE;
}

/*
 * Function: lookup_read_num_directory_entry_handles
 *
 * Synopsis: Given a directory handle, determine how many handles are
 * used to store directory entries for this directory.
 *
 * Initializes key and value structures to direct num handles into
 * s_op->u.lookup.num_dirent_handles. The handle to use for the
 * read is either:
 * - the starting handle from the req (if we haven't looked up a
 *   segment yet), or
 * - the previous segment's handle (from response handle array).
 *
 * Posts the keyval read to trove.
 */
static PINT_sm_action lookup_read_num_directory_entry_handles(
        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;
    PVFS_handle handle = PVFS_HANDLE_NULL;
    job_id_t j_id;

    /* use the base handle if we haven't looked up a segment yet */
    if (s_op->u.lookup.seg_nr == 0)
    {
        handle = s_op->req->u.lookup_path.handle;
    }
    else
    {
        handle = s_op->resp.u.lookup_path.handle_array[
            s_op->u.lookup.seg_nr-1];
    }

    gossip_debug(GOSSIP_SERVER_DEBUG,
                 "  reading distributed directory attributes from handle %llu\n",
                 llu(handle));

    s_op->key.buffer = Trove_Common_Keys[DIST_DIR_ATTR_KEY].key;
    s_op->key.buffer_sz = Trove_Common_Keys[DIST_DIR_ATTR_KEY].size;
    s_op->val.buffer = &s_op->u.lookup.attr.u.dir.dist_dir_attr;
    s_op->val.buffer_sz = sizeof(s_op->u.lookup.attr.u.dir.dist_dir_attr);

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

    return ret;
}

/*
 * Function: lookup_read_directory_entry_handles
 *
 * Synopsis: Given a directory handle, look up the handle used to
 * store directory entries for this directory.
 *
 * Initializes key and value structures to direct handle into
 * s_op->u.lookup.dirent_handle, which is where we always store the
 * handle used to read directory entries.  The handle to use for the
 * read is either:
 * - the starting handle from the req (if we haven't looked up a
 *   segment yet), or
 * - the previous segment's handle (from response handle array).
 *
 * Posts the keyval read to trove.
 */
static PINT_sm_action lookup_read_directory_entry_handles(
        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;
    PVFS_handle handle = PVFS_HANDLE_NULL;
    job_id_t j_id;
    PVFS_object_attr *attr;
    int keyval_count;

    attr = &s_op->u.lookup.attr;

    if(js_p->error_code == -TROVE_ENOENT)
    {
        gossip_debug(GOSSIP_GETATTR_DEBUG, "lookup: no DIST_DIR_ATTR key present!\n");
        attr->u.dir.dist_dir_bitmap = NULL;
        attr->u.dir.dirdata_handles = NULL;
        return SM_ACTION_COMPLETE;
    }

    assert(attr->u.dir.dist_dir_attr.num_servers > 0 &&
        attr->u.dir.dist_dir_attr.bitmap_size > 0);

    /* use the base handle if we haven't looked up a segment yet */
    if (s_op->u.lookup.seg_nr == 0)
    {
        handle = s_op->req->u.lookup_path.handle;
    }
    else
    {
        handle = s_op->resp.u.lookup_path.handle_array[
            s_op->u.lookup.seg_nr-1];
    }

    gossip_debug(GOSSIP_SERVER_DEBUG,
            "lookup: get dist-dir-attr for dir meta handle %llu "
            "with tree_height=%d, num_servers=%d, bitmap_size=%d, "
            "split_size=%d, server_no=%d and branch_level=%d\n",
            llu(handle),
            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);

    /* allocate space for bitmap and dirdata handles */
    attr->u.dir.dist_dir_bitmap =
        malloc(attr->u.dir.dist_dir_attr.bitmap_size *
                sizeof(PVFS_dist_dir_bitmap_basetype));
    attr->u.dir.dirdata_handles =
        malloc(attr->u.dir.dist_dir_attr.num_servers *
                sizeof(PVFS_handle));
    if(!attr->u.dir.dist_dir_bitmap ||
            !attr->u.dir.dirdata_handles)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }


    /* total 2 keyvals, DIST_DIRDATA_BITMAP, DIST_DIRDATA_HANDLES */
    keyval_count = 2;

    s_op->key_a = malloc(sizeof(PVFS_ds_keyval) * keyval_count);
    if(!s_op->key_a)
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }

    s_op->val_a = malloc(sizeof(PVFS_ds_keyval) * keyval_count);
    if(!s_op->val_a)
    {
        free(s_op->key_a);
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }
    memset(s_op->val_a, 0, sizeof(PVFS_ds_keyval) * keyval_count);

    s_op->key_a[0].buffer = Trove_Common_Keys[DIST_DIRDATA_BITMAP_KEY].key;
    s_op->key_a[0].buffer_sz = Trove_Common_Keys[DIST_DIRDATA_BITMAP_KEY].size;

    s_op->val_a[0].buffer_sz =
        attr->u.dir.dist_dir_attr.bitmap_size *
        sizeof(PVFS_dist_dir_bitmap_basetype);
    s_op->val_a[0].buffer = attr->u.dir.dist_dir_bitmap;

    s_op->key_a[1].buffer = Trove_Common_Keys[DIST_DIRDATA_HANDLES_KEY].key;
    s_op->key_a[1].buffer_sz = Trove_Common_Keys[DIST_DIRDATA_HANDLES_KEY].size;

    s_op->val_a[1].buffer = attr->u.dir.dirdata_handles;
    s_op->val_a[1].buffer_sz = attr->u.dir.dist_dir_attr.num_servers *
        sizeof(PVFS_handle);

    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,
        keyval_count,
        0,
        NULL,
        smcb,
        0,
        js_p,
        &j_id,
        server_job_context, s_op->req->hints);

    return ret;
}

/*
 * Function: lookup_read_directory_entry
 *
 * Synopsis: Given a handle for a dspace holding directory entries,
 * look up the current segment and obtain its handle.
 */
static PINT_sm_action lookup_read_directory_entry(
        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 j_id;

    /* TODO: Need to see if directory entry is on this server. If not we
       can't continue processing the current segment. */
    gossip_debug(
        GOSSIP_SERVER_DEBUG, "  need to read from dirent handle = "
        "%llu, segment = %s (len=%d)\n", llu(s_op->u.lookup.attr.u.dir.dirdata_handles[0]),
        s_op->u.lookup.segp, (int) strlen(s_op->u.lookup.segp));

    s_op->key.buffer = s_op->u.lookup.segp;
    s_op->key.buffer_sz = strlen(s_op->u.lookup.segp) + 1;
    s_op->val.buffer =
        &s_op->resp.u.lookup_path.handle_array[s_op->u.lookup.seg_nr];
    s_op->val.buffer_sz = sizeof(PVFS_handle);

    /*
      NOTE: if this operation fails, seg_nr will indicate one too many
      valid segments; this is addressed in lookup_send_response.
    */
    s_op->u.lookup.seg_nr++;

    ret = job_trove_keyval_read(
        s_op->req->u.lookup_path.fs_id, s_op->u.lookup.attr.u.dir.dirdata_handles[0],
        &s_op->key, &s_op->val, 
        0, 
        NULL, smcb, 0, js_p, &j_id,
        server_job_context, s_op->req->hints);

    return ret;
}


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

    /*
      NOTE: we may have handle_count N with attr_count N-1 in the case
      that another meta-server needs to continue with the path (attr)
      lookup.  otherwise, we're returning N handles with N attrs.

      # actually completed are one less than the respective counts
    */
    s_op->resp.u.lookup_path.handle_count = s_op->u.lookup.handle_ct - 1;
    s_op->resp.u.lookup_path.attr_count = s_op->u.lookup.attr_ct - 1;

    if (s_op->resp.u.lookup_path.handle_count ||
        s_op->resp.u.lookup_path.attr_count)
    {
        js_p->error_code = 0;
    }
    else if(js_p->error_code < 0)
    {
        /* preserve error code in this case and fall through */
        gossip_debug(GOSSIP_SERVER_DEBUG, "  lookup error in previous step\n");
    }
    else
    {
        js_p->error_code = -PVFS_ENOENT;
    }

    gossip_debug(GOSSIP_SERVER_DEBUG, "  sending '%s' response with %d "
                 "handle(s) and %d attr(s)\n",
                 (js_p->error_code ? "error" : "success"),
                 s_op->resp.u.lookup_path.handle_count,
                 s_op->resp.u.lookup_path.attr_count);

    if(js_p->error_code == 0)
    {
        PINT_ACCESS_DEBUG(
            s_op, GOSSIP_ACCESS_DEBUG, "path: %s, handle: %llu\n", 
            s_op->req->u.lookup_path.path, 
            llu(s_op->resp.u.lookup_path.handle_array[
                s_op->resp.u.lookup_path.handle_count-1]));
    }
    else
    {
        PINT_ACCESS_DEBUG(
            s_op, GOSSIP_ACCESS_DEBUG, "path: %s, lookup failed\n", 
            s_op->req->u.lookup_path.path);
    }

    return SM_ACTION_COMPLETE;
}

/*
 * Function: lookup_cleanup
 *
 * Synopsis: Free memory allocated during request processing.
 *
 * There are a bunch of regions that must be freed after processing
 * completes:
 * - decoded request (s_op->decoded)
 * - encoded request (s_op->unexp_bmi_buff.buffer)
 * - encoded response (s_op->encoded)
 * - original (decoded) response (s_op->resp)
 * - dynamically allocated space (in this case 
 *   s_op->resp.u.lookup_path.handle_array)
 * - the server operation structure itself
 */
static PINT_sm_action lookup_cleanup(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    if (s_op->resp.u.lookup_path.handle_array)
    {
        free(s_op->resp.u.lookup_path.handle_array);
        s_op->resp.u.lookup_path.handle_array = NULL;
    }

    if (s_op->u.lookup.ds_attr_array)
    {
        free(s_op->u.lookup.ds_attr_array);
        s_op->u.lookup.ds_attr_array = NULL;
    }

    /* TODO: Need to free bitmap and handles arrays. */

    return(server_state_machine_complete(smcb));
}

PINT_GET_OBJECT_REF_DEFINE(lookup_path);

struct PINT_server_req_params pvfs2_lookup_params =
{
    .string_name = "lookup_path",
    .perm = PINT_SERVER_CHECK_NONE,
    .sched_policy = PINT_SERVER_REQ_SCHEDULE,
    .get_object_ref = PINT_get_object_ref_lookup_path,
    .state_machine = &pvfs2_lookup_sm
};

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