root/branches/Orange-Elaine-Distr-Dir-Branch/src/server/mkdir.sm @ 8648

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


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

#include "server-config.h"
#include "pvfs2-server.h"
#include "pvfs2-attr.h"
#include "pvfs2-util.h"
#include "pvfs2-internal.h"
#include "pint-util.h"
#include "pint-cached-config.h"
#include "pvfs2-dist-basic.h"
#include "dist-dir-utils.h"

%%

nested machine pvfs2_mkdir_work_sm
{
    state create
    {
        run mkdir_create;
        success => set_attrib;
        default => err_msg;
    }

    state set_attrib
    {
        run mkdir_setattrib;
        success => create_dist_dir_struct;
        default => err_msg;
    }

/* 1. now use err_msg as the error handling, will change if necessary */

    state create_dist_dir_struct
    {
        run mkdir_create_dist_dir_struct;
        success => select_dirdata_servers;
        default => err_msg;
    }

    state select_dirdata_servers
    {
        run mkdir_select_dirdata_servers;
        success => create_local_dirdata_dspace;
        default => err_msg;
    }

    state create_local_dirdata_dspace
    {
        run mkdir_create_local_dirdata_dspace;
        success => request_remote_dirdata_dspace;
        default => err_msg;
    }

    state request_remote_dirdata_dspace
    {
        run mkdir_request_remote_dirdata_dspace;
        success => fill_handles_to_dist_dir_struct;
        default => err_msg;
    }

    state fill_handles_to_dist_dir_struct
    {
        run mkdir_fill_handles_to_dist_dir_struct;
        default => write_keyval_to_meta_handle;
/*
        default => remote_dirdata_attr_and_keyval_setup_msgpair;
*/
    }

/*
    state remote_dirdata_attr_and_keyval_setup_msgpair
    {
        run mkdir_remote_dirdata_attr_and_keyval_setup_msgpair;
        success => remote_dirdata_attr_and_keyval_xfer_msgpair;
        default => err_msg;
    }

    state remote_dirdata_attr_and_keyval_xfer_msgpair
    {
        jump pvfs2_msgpairarray_sm;
        success => write_keyval_to_meta_handle;
        default => err_msg;
    }
*/
    
    state write_keyval_to_meta_handle
    {
        run mkdir_write_keyval_to_meta_handle;
        success => return;
        default => err_msg;
    }

    state err_msg 
    {
        run mkdir_error;
        default => return;
    }
}

machine pvfs2_mkdir_sm
{
    state prelude
    {
        jump pvfs2_prelude_sm;
        success => prep_sm;
        default => final_response;
    }

    state prep_sm
    {
        run mkdir_prep_sm;
        default => work;
    }

    state work
    {
        jump pvfs2_mkdir_work_sm;
        default => final_response;
    }

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

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

%%

/*
 * Function: mkdir_create
 *
 * Params:   server_op *s_op, 
 *           job_status_s *js_p
 *
 * Returns:  int
 *
 * Synopsis: 
 *
 * NOTE: returned handle will pop out in js_p->handle (the job status
 * struct).
 */
static PINT_sm_action mkdir_create(
        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;

    gossip_debug(GOSSIP_MKDIR_DEBUG, " creating dspace on coll_id %d\n",
                 s_op->u.mkdir.fs_id);

    gossip_debug(
        GOSSIP_MKDIR_DEBUG, " directory handle will "
        "be in starting hrange[0] %llu-%llu\n",
        llu(s_op->u.mkdir.handle_extent_array.extent_array[0].first),
        llu(s_op->u.mkdir.handle_extent_array.extent_array[0].last));

    ret = job_trove_dspace_create(
        s_op->u.mkdir.fs_id, &s_op->u.mkdir.handle_extent_array,
        PVFS_TYPE_DIRECTORY, NULL,
        TROVE_SYNC, 
        smcb, 0, js_p, &i,
        server_job_context, s_op->req->hints);

    return ret;
}

/*
 * Function: mkdir_setattrib
 *
 * Params:   server_op *s_op, 
 *           job_status_s *js_p
 *
 * Returns:  int
 *
 * Synopsis: 
 *           
 */
static PINT_sm_action mkdir_setattrib(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -1;
    job_id_t j_id;
    PVFS_ds_attributes *ds_attr = NULL;

    /* save the handle from the mkdir in the response */
    s_op->resp.u.mkdir.handle = js_p->handle;

    gossip_debug(GOSSIP_MKDIR_DEBUG, " *** new directory has handle %llu\n",
                 llu(js_p->handle));

    ds_attr = (&s_op->ds_attr);
    PVFS_ds_init_time(ds_attr);

    gossip_debug(GOSSIP_MKDIR_DEBUG, " setting directory version to "
                 "%llu\n\tmtime is %llu\n", llu(ds_attr->mtime),
                 llu(PINT_util_mkversion_time(ds_attr->mtime)));

    ret = job_trove_dspace_setattr(
        s_op->u.mkdir.fs_id, s_op->resp.u.mkdir.handle,
        ds_attr, 
        TROVE_SYNC,
        smcb, 0, js_p, &j_id, server_job_context, s_op->req->hints);

    return ret;
}


/* init dist_dir_struct in s_op->attr.u.dir and s_op->u.mkdir */
static PINT_sm_action mkdir_create_dist_dir_struct(
        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 *attr;
    int num_total_dirdata_servers, num_initial_dirdata_servers;
    server_configuration_s *user_opts = get_server_config_struct();
    int ret = -1;
    unsigned char *c;
    int i;

    attr = (&s_op->attr);

    /* use total # of meta servers as the num_total_dirdata_servers */
    ret = PINT_cached_config_get_num_meta(s_op->u.mkdir.fs_id,
                                            &num_total_dirdata_servers);
    if(ret < 0)
    {
        gossip_err("Error: failed to get number of metadata servers\n");
        js_p->error_code = ret;
        return SM_ACTION_COMPLETE;
    }

    /* set num_dirdata_servers for pre-allocation of dirdata handles */
    s_op->u.mkdir.num_dirdata_servers =
        num_total_dirdata_servers;

    /* TODO: if received >0, use; else use config_file value,*/
    if(s_op->req->u.mkdir.num_dirent_files_req > 0)
    {
        num_initial_dirdata_servers = s_op->req->u.mkdir.num_dirent_files_req;
    }
    else
    {
        num_initial_dirdata_servers = user_opts->init_num_dirdata_handles;
    }

    /* will do the same check in PINT_init_dist_dir_state, 
     * but can also add here */
    if(num_initial_dirdata_servers > num_total_dirdata_servers)
    {
        num_initial_dirdata_servers = num_total_dirdata_servers;
    }

    /* init meta handle dist_dir_struct */
    ret = PINT_init_dist_dir_state(&attr->u.dir.dist_dir_attr,
            &attr->u.dir.dist_dir_bitmap,
            num_total_dirdata_servers,
            0,
            num_initial_dirdata_servers);

    assert(ret == 0);

    gossip_debug(GOSSIP_MKDIR_DEBUG, 
            "mkdir: Init 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(s_op->resp.u.mkdir.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);

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

    /* allocate dirdata handle space, similar to create.sm */
    attr->u.dir.dirdata_handles = malloc(
            sizeof(*attr->u.dir.dirdata_handles) *
            num_total_dirdata_servers);
    s_op->u.mkdir.handle_array_local = malloc(
            sizeof(*s_op->u.mkdir.handle_array_local) *
            num_total_dirdata_servers);
    s_op->u.mkdir.handle_array_remote = malloc(
            sizeof(*s_op->u.mkdir.handle_array_remote) *
            num_total_dirdata_servers);

    if(!attr->u.dir.dirdata_handles ||
            !s_op->u.mkdir.handle_array_local ||
            !s_op->u.mkdir.handle_array_remote )
    {
        js_p->error_code = -PVFS_ENOMEM;
        return SM_ACTION_COMPLETE;
    }

    return SM_ACTION_COMPLETE; 
}


static PINT_sm_action mkdir_select_dirdata_servers(
        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 *attr;
    */

/* not selecting servers now, suppose has dirdata on every metadata server */

        /******** if use the function in create.sm to get server list:
*
int PINT_cached_config_get_server_list(
    PVFS_fs_id fs_id,
    PINT_dist *dist,
    int num_dfiles_req,
    PVFS_sys_layout *layout,
    const char ***server_names,
    int *server_count)

1. need to provide dist & layout
2. server_count might be different from num_dfiles_req, then better put create dist-dir-struct after the servers selection function.

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


        return SM_ACTION_COMPLETE;

}



static PINT_sm_action mkdir_create_local_dirdata_dspace(
        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 *attr;
    int ret = -PVFS_ENOMEM;
    job_id_t tmp_id;
    PVFS_handle_extent_array extent_array;
    server_configuration_s *user_opts = get_server_config_struct();

    attr = (&s_op->attr);

    assert(s_op->u.mkdir.num_dirdata_servers > 0);

    /* dirdata on every meta server */
    s_op->u.mkdir.handle_array_local_count = 1;
    s_op->u.mkdir.handle_array_remote_count = 
        s_op->u.mkdir.num_dirdata_servers - 1;

    gossip_debug(GOSSIP_MKDIR_DEBUG, "creating %d local dirdata files\n", 
        s_op->u.mkdir.handle_array_local_count);
    gossip_debug(GOSSIP_MKDIR_DEBUG, "creating %d remote dirdata files\n", 
        s_op->u.mkdir.handle_array_remote_count);

    if(s_op->u.mkdir.handle_array_local_count == 0)
    {
        /* no local work to do */
        js_p->error_code = 0;
        return(SM_ACTION_COMPLETE);
    }

    ret = PINT_config_get_meta_handle_extent_array(
        user_opts, s_op->u.mkdir.fs_id, &extent_array);
    assert(ret == 0);
    assert(extent_array.extent_count > 0);

    gossip_debug(GOSSIP_MKDIR_DEBUG, " local dirdata handle(s) will "
                 "be in starting hrange[0] %llu-%llu\n",
                 llu(extent_array.extent_array[0].first),
                 llu(extent_array.extent_array[0].last));

    ret = job_trove_dspace_create_list(
        s_op->req->u.mkdir.fs_id,
        &extent_array,
        s_op->u.mkdir.handle_array_local,
        s_op->u.mkdir.handle_array_local_count,
        PVFS_TYPE_DIRDATA, 
        NULL,
        TROVE_SYNC,
        smcb,
        0,
        js_p,
        &tmp_id,
        server_job_context, 
        s_op->req->hints);

    return(ret);
}


static PINT_sm_action mkdir_request_remote_dirdata_dspace(
        struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    int ret = -1;
    job_id_t j_id;

    gossip_debug(
        GOSSIP_MKDIR_DEBUG, "  local dirdata handle (%llu)\n",
        llu(s_op->u.mkdir.handle_array_local[0]));

    if(s_op->u.mkdir.handle_array_remote_count == 0)
    {
        js_p->error_code = 0;
        return(SM_ACTION_COMPLETE);
    }

    ret = job_precreate_pool_get_handles(
        s_op->req->u.mkdir.fs_id,
        s_op->u.mkdir.handle_array_remote_count,
        PVFS_TYPE_DIRDATA,
        NULL, /* server string set to NULL for now */
        s_op->u.mkdir.handle_array_remote,
        0,
        smcb,
        0,
        js_p,
        &j_id,
        server_job_context,
        s_op->req->hints);

    return ret;

}


static PINT_sm_action mkdir_fill_handles_to_dist_dir_struct(
        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 *attr;
    int i;

    attr = &s_op->attr;

    /* fill dirdata_handles array */
    attr->u.dir.dirdata_handles[0] =
        s_op->u.mkdir.handle_array_local[0];

    for(i = 0; i < s_op->u.mkdir.num_dirdata_servers - 1; i++)
    {
        attr->u.dir.dirdata_handles[i+1] =
            s_op->u.mkdir.handle_array_remote[i];
    }

    gossip_debug(GOSSIP_MKDIR_DEBUG, 
            "\t mkdir: dirdata handles array of directory %llu.\n",
            llu(s_op->resp.u.mkdir.handle));

    for(i=0; i<s_op->u.mkdir.num_dirdata_servers; i++)
    {
        gossip_debug(GOSSIP_MKDIR_DEBUG, 
                "\t\tdirdata server %d: %llu.\n",
                i, llu(attr->u.dir.dirdata_handles[i]));
    }

    return SM_ACTION_COMPLETE;
}

#if 0
static int mkdir_remote_dirdata_attr_and_keyval_setup_msgpair(struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    PINT_sm_msgpair_state *msg_p = NULL;
    PVFS_object_attr attr;
    int ret;

    gossip_debug(GOSSIP_MKDIR_DEBUG, "mkdir state: send_remote_dir_data\n");

    /* !!! set dirdata attr, set all the static component */
    memcpy(&attr, &s_op->req->u.mkdir.attr, sizeof(attr));
    attr.objtype = PVFS_TYPE_DIRDATA;
    PINT_dist_dir_attr_copyto(attr.u.dir.dist_dir_attr,
                              s_op->attr.u.dir.dist_dir_attr);
    attr.u.dir.dirdata_handles = s_op->attr.u.dir.dirdata_handles;
    attr.u.dir.dist_dir_bitmap = s_op->attr.u.dir.dist_dir_bitmap;
    attr.mask = PVFS_ATTR_COMMON_ALL;
    /* not writing DISTDIR_ATTR */ 
    /*
       attr.mask |= PVFS_ATTR_DIR_DISTDIR_ATTR;
     */

    gossip_debug(
        GOSSIP_MKDIR_DEBUG,
        "  SENDING attrs: [owner = %d, group = %d\n\t"
        "perms = %o, type = %d, atime = %llu, mtime = %llu\n\t"
        "ctime = %llu ]\n",
        attr.owner, attr.group, attr.perms,
        attr.objtype, llu(attr.atime),
        llu(PINT_util_mkversion_time(attr.mtime)), llu(attr.ctime));

    PINT_msgpair_init(&s_op->msgarray_op);
    msg_p = &s_op->msgarray_op.msgpair;
    PINT_serv_init_msgarray_params(s_op, s_op->u.mkdir.fs_id);

    PINT_SERVREQ_TREE_SETATTR_FILL(
        msg_p->req,
        s_op->req->credentials,
        s_op->u.mkdir.fs_id,
        PVFS_TYPE_DIRDATA,
        attr,
        s_op->attr.u.dir.dist_dir_attr.num_servers,
        s_op->attr.u.dir.dirdata_handles,
        NULL);

    msg_p->fs_id = s_op->u.mkdir.fs_id;
    msg_p->handle = s_op->attr.u.dir.dirdata_handles[0];
    msg_p->retry_flag = PVFS_MSGPAIR_RETRY;
    msg_p->comp_fn = NULL;

    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 dirdata server address\n");
        js_p->error_code = ret;
    }

    PINT_sm_push_frame(smcb, 0, &s_op->msgarray_op);
    js_p->error_code = 0;
    return SM_ACTION_COMPLETE;
}
#endif

static PINT_sm_action mkdir_write_keyval_to_meta_handle(
        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 *attr;
    int ret = -1;
    job_id_t j_id;

    /* total 3 keyvals, PVFS_DIST_DIR_ATTR, PVFS_DIRDATA_BITMAP, PVFS_DIRDATA_HANDLES */
    int keyval_count = 3;

    /* !!!! this is to set meta handle keyval,
     * use dist_dir_attr in s_op->attr 
     */
    attr = &s_op->attr;

    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_DIR_ATTR_KEY].key;
    s_op->key_a[0].buffer_sz = Trove_Common_Keys[DIST_DIR_ATTR_KEY].size;

    s_op->val_a[0].buffer = &attr->u.dir.dist_dir_attr;
    s_op->val_a[0].buffer_sz =
         sizeof(attr->u.dir.dist_dir_attr);

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

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

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

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

    gossip_debug(GOSSIP_MKDIR_DEBUG, 
            "  writing dist-dir-struct keyvals for meta handle: %llu "
            "\t with server_no=%d and branch_level=%d \n", 
            llu(s_op->resp.u.mkdir.handle),
            attr->u.dir.dist_dir_attr.server_no,
            attr->u.dir.dist_dir_attr.branch_level);


    ret = job_trove_keyval_write_list(
            s_op->req->u.mkdir.fs_id,
            s_op->resp.u.mkdir.handle, /* meta handle */
            s_op->key_a, s_op->val_a,
            keyval_count, TROVE_SYNC, NULL, smcb,
            0, js_p, &j_id, server_job_context,
            s_op->req->hints);

    return ret;
}

static int mkdir_error(struct PINT_smcb *smcb, job_status_s *js_p)
{
    struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
    s_op->resp.u.mkdir.handle = 0;

    gossip_debug(
        GOSSIP_MKDIR_DEBUG, "  mkdir_error: error_code = %d\n",
        js_p->error_code);

    if (js_p->error_code > -1)
    {
        js_p->error_code = -PVFS_EINVAL;
    }
    return 1;
}


/*
 * Function: mkdir_cleanup
 *
 * Params:   server_op *b, 
 *           job_status_s *js_p
 *
 * Returns:  int
 *
 * Synopsis: free memory and return
 *           
 */
static PINT_sm_action mkdir_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->key_a)
    {
        free(s_op->key_a);
    }

    if(s_op->val_a)
    {
        free(s_op->val_a);
    }

    PINT_free_object_attr(&s_op->attr);

    if(s_op->u.mkdir.handle_array_local)
    {
        free(s_op->u.mkdir.handle_array_local);
    }

    if(s_op->u.mkdir.handle_array_remote)
    {
        free(s_op->u.mkdir.handle_array_remote);
    }

    return(server_state_machine_complete(smcb));
}

static PINT_sm_action mkdir_prep_sm(
        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 *a_p = NULL;
    PVFS_ds_attributes *ds_attr = NULL;

    s_op->u.mkdir.fs_id = s_op->req->u.mkdir.fs_id;
    s_op->u.mkdir.handle_extent_array =
        s_op->req->u.mkdir.handle_extent_array;

    a_p = &(s_op->req->u.mkdir.attr);

    if (a_p->objtype != PVFS_TYPE_DIRECTORY)
    {
        js_p->error_code = -PVFS_EINVAL;
        return SM_ACTION_COMPLETE;
    }

    gossip_debug(GOSSIP_MKDIR_DEBUG, " attrs to write: [owner = %d, "
                 "group = %d, perms = %o]\n", a_p->owner, a_p->group,
                 a_p->perms);

    /* translate attrs to storage attr format */
    ds_attr = (&s_op->ds_attr);
    PVFS_object_attr_to_ds_attr(a_p, ds_attr);

    js_p->error_code = 0;
    return SM_ACTION_COMPLETE;
}

static inline int PINT_get_object_ref_mkdir(
    struct PVFS_server_req *req, PVFS_fs_id *fs_id, PVFS_handle *handle)
{
    *fs_id = req->u.mkdir.fs_id;
    *handle = PVFS_HANDLE_NULL;
    return 0;
};

struct PINT_server_req_params pvfs2_mkdir_params =
{
    .string_name = "mkdir",
    .perm = PINT_SERVER_CHECK_NONE,
    .access_type = PINT_server_req_modify,
    .state_machine = &pvfs2_mkdir_sm
};

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