home *** CD-ROM | disk | FTP | other *** search
/ ftp.parl.clemson.edu / 2015-02-07.ftp.parl.clemson.edu.tar / ftp.parl.clemson.edu / pub / pvfs2 / orangefs-2.8.3-20110323.tar.gz / orangefs-2.8.3-20110323.tar / orangefs / src / server / create.sm < prev    next >
Text File  |  2011-03-22  |  25KB  |  880 lines
  1. /* 
  2.  * (C) 2001 The University of Chicago 
  3.  *
  4.  * See COPYING in top-level directory.
  5.  */
  6.  
  7. #include <string.h>
  8. #include <assert.h>
  9.  
  10. #include "server-config.h"
  11. #include "pvfs2-server.h"
  12. #include "pvfs2-attr.h"
  13. #include "gossip.h"
  14. #include "pvfs2-internal.h"
  15. #include "pint-util.h"
  16. #include "pint-cached-config.h"
  17.  
  18. #define REPLACE_DONE 100
  19.  
  20. %%
  21.  
  22. machine pvfs2_create_sm
  23. {
  24.     state prelude
  25.     {
  26.         jump pvfs2_prelude_sm;
  27.         success => create_metafile;
  28.         default => setup_final_response;
  29.     }
  30.  
  31.     state create_metafile
  32.     {
  33.         run create_metafile;
  34.         success => check_stuffed;
  35.         default => setup_final_response;
  36.     }
  37.  
  38.     state check_stuffed
  39.     {
  40.         run check_stuffed;
  41.         success => create_local_datafiles;
  42.         default => setup_final_response;
  43.     }
  44.  
  45.     state create_local_datafiles
  46.     {
  47.         run create_local_datafiles;
  48.         success => setup_local_datafile_handles;
  49.         default => remove_metafile_object;
  50.     }
  51.  
  52.     state setup_local_datafile_handles
  53.     {
  54.         run setup_local_datafile_handles;
  55.         success => request_datafiles;
  56.         default => remove_local_datafile_handles;
  57.     }
  58.  
  59.     state request_datafiles
  60.     {
  61.         run request_datafiles;
  62.         success => write_keyvals;
  63.         default => remove_local_datafile_handles;
  64.     }
  65.  
  66.     state write_keyvals
  67.     {
  68.         run write_keyvals;
  69.         success => setobj_attribs;
  70.         default => replace_remote_datafile_handles;
  71.     }
  72.  
  73.     state setobj_attribs
  74.     {
  75.         run setattr_setobj_attribs;
  76.         success => setup_resp;
  77.         default => remove_keyvals;
  78.     }
  79.  
  80.     state setup_resp
  81.     {
  82.         run setup_resp;
  83.         default => setup_final_response;
  84.     }
  85.  
  86.     state remove_local_datafile_handles
  87.     {
  88.         run remove_local_datafile_handles;
  89.         default => remove_metafile_object;
  90.     }
  91.  
  92.     state replace_remote_datafile_handles
  93.     {
  94.         run replace_remote_datafile_handles;
  95.         REPLACE_DONE => remove_local_datafile_handles;
  96.         default => replace_remote_datafile_handles;
  97.     }
  98.  
  99.     state remove_metafile_object
  100.     {
  101.         run remove_metafile_object;
  102.         default => setup_final_response;
  103.     }
  104.  
  105.     state remove_keyvals
  106.     {
  107.         run remove_keyvals;
  108.         success => replace_remote_datafile_handles;
  109.         default => setup_final_response;
  110.     }
  111.  
  112.     state setup_final_response
  113.     {
  114.         run setup_final_response;
  115.         default => final_response;
  116.     }
  117.  
  118.     state final_response
  119.     {
  120.         jump pvfs2_final_response_sm;
  121.         default => cleanup;
  122.     }
  123.  
  124.     state cleanup
  125.     {
  126.         run cleanup;
  127.         default => terminate;
  128.     }
  129. }
  130.  
  131. %%
  132.  
  133. static int setup_final_response(
  134.     struct PINT_smcb *smcb, job_status_s *js_p)
  135. {
  136.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  137.  
  138.     /* retrieve original error code if present */
  139.     if(s_op->u.create.saved_error_code)
  140.     {
  141.         js_p->error_code = s_op->u.create.saved_error_code;
  142.     }
  143.  
  144.     /* otherwise propigate the js_p->error code */
  145.     return(SM_ACTION_COMPLETE);
  146. }
  147.  
  148. static int create_metafile(
  149.         struct PINT_smcb *smcb, job_status_s *js_p)
  150. {
  151.  
  152.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  153.     int ret = -1;
  154.     job_id_t i;
  155.     PVFS_handle_extent_array meta_handle_ext_array;
  156.     server_configuration_s *config = get_server_config_struct();
  157.  
  158.     /* first state to check in, make sure the attr mask contains the dist bit. 
  159.      * it's required later (not sure if we have to require it) but if we don't
  160.      * have it here, return an EINVAL */
  161.     if( ! (s_op->req->u.create.attr.mask & PVFS_ATTR_META_DIST) )
  162.     {
  163.         gossip_debug(GOSSIP_SERVER_DEBUG, "%s: invalid create request, "
  164.                      "attribute mask did not include the distribution\n",
  165.                      __func__);
  166.         js_p->error_code = -PVFS_EINVAL;
  167.         return SM_ACTION_COMPLETE;
  168.     }
  169.  
  170.     ret = PINT_cached_config_get_server(
  171.         s_op->req->u.create.fs_id,
  172.         config->host_id,
  173.         PINT_SERVER_TYPE_META,
  174.         &meta_handle_ext_array);
  175.  
  176.     ret = job_trove_dspace_create(
  177.         s_op->req->u.create.fs_id,
  178.         &meta_handle_ext_array,
  179.         PVFS_TYPE_METAFILE,
  180.         NULL,
  181.         0,
  182.         smcb,
  183.         0,
  184.         js_p,
  185.         &i,
  186.         server_job_context, s_op->req->hints);
  187.  
  188.     return(ret);
  189. }
  190.  
  191. static int check_stuffed(
  192.     struct PINT_smcb *smcb, job_status_s *js_p)
  193. {
  194.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  195.     int server_type;
  196.     server_configuration_s *config = get_server_config_struct();
  197.     struct filesystem_configuration_s *fs_conf;
  198.     PVFS_BMI_addr_t myaddr;
  199.     PVFS_sys_layout *layout;
  200.     int ret;
  201.     const char* svr_name;
  202.     int i;
  203.  
  204.     s_op->resp.u.create.metafile_handle = js_p->handle;
  205.     gossip_debug(
  206.         GOSSIP_SERVER_DEBUG, "Metafile handle created: %llu\n",
  207.         llu(js_p->handle));
  208.  
  209.     assert(config);
  210.  
  211.     layout = &s_op->req->u.create.layout;
  212.  
  213.     if(layout->algorithm == PVFS_SYS_LAYOUT_LIST)
  214.     {
  215.         for(i=0; i<layout->server_list.count; i++)
  216.         {
  217.             gossip_debug(GOSSIP_SERVER_DEBUG, "layout list server %d: %lld\n", 
  218.                 i, lld(layout->server_list.servers[i])); 
  219.         }
  220.     }
  221.  
  222.     fs_conf = PINT_config_find_fs_id(config, 
  223.         s_op->req->u.create.fs_id);
  224.     if(!fs_conf)
  225.     {
  226.         js_p->error_code = -PVFS_EINVAL;
  227.         return(SM_ACTION_COMPLETE);
  228.     }
  229.  
  230.     ret = BMI_addr_lookup(&myaddr, config->host_id);
  231.     if(ret != 0)
  232.     {
  233.         /* we can't get our own address? */
  234.         js_p->error_code = ret;
  235.         return SM_ACTION_COMPLETE;
  236.     }
  237.  
  238.     /* is this metadata server also IO? */
  239.     svr_name = PINT_cached_config_map_addr(s_op->req->u.create.fs_id,
  240.                                 myaddr, &server_type);
  241.     if(!svr_name)
  242.     {
  243.         js_p->error_code = ret;
  244.         return SM_ACTION_COMPLETE;
  245.     }
  246.  
  247.     /* for now only support stuffing of ROUND_ROBIN layouts */
  248.     if((server_type & PINT_SERVER_TYPE_IO) && fs_conf->file_stuffing && layout->algorithm == PVFS_SYS_LAYOUT_ROUND_ROBIN)
  249.     {    
  250.         /* we can do a stuffed create here, only one datafile */
  251.         s_op->req->u.create.attr.u.meta.dfile_count = 1;
  252.         s_op->resp.u.create.datafile_count = 1;
  253.         s_op->resp.u.create.datafile_handles = malloc(sizeof(PVFS_handle));
  254.         s_op->u.create.handle_array_local = malloc(sizeof(PVFS_handle));
  255.         if(!s_op->resp.u.create.datafile_handles || !s_op->u.create.handle_array_local)
  256.         {
  257.             js_p->error_code = -PVFS_ENOMEM;
  258.             return SM_ACTION_COMPLETE;
  259.         }
  260.  
  261.         s_op->resp.u.create.stuffed = 1;
  262.         s_op->u.create.handle_array_local[0] = 100;
  263.         js_p->error_code = 0;
  264.         return SM_ACTION_COMPLETE;
  265.     }
  266.  
  267.     /* file will not be stuffed; need to allocate all datafiles */
  268.     s_op->u.create.num_io_servers = s_op->req->u.create.num_dfiles_req;
  269.     s_op->resp.u.create.datafile_handles = malloc(
  270.         sizeof(*s_op->resp.u.create.datafile_handles) *
  271.         s_op->u.create.num_io_servers);
  272.     s_op->u.create.handle_array_local = malloc(
  273.         sizeof(*s_op->u.create.handle_array_local) *
  274.         s_op->u.create.num_io_servers);
  275.     s_op->u.create.handle_array_remote = malloc(
  276.         sizeof(*s_op->u.create.handle_array_remote) *
  277.         s_op->u.create.num_io_servers);
  278.     s_op->u.create.remote_io_servers = malloc(
  279.         sizeof(char*) *
  280.         s_op->u.create.num_io_servers);
  281.     if(!s_op->resp.u.create.datafile_handles || 
  282.         !s_op->u.create.handle_array_local ||
  283.         !s_op->u.create.handle_array_remote ||
  284.         !s_op->u.create.remote_io_servers)
  285.     {
  286.         js_p->error_code = -PVFS_ENOMEM;
  287.         return SM_ACTION_COMPLETE;
  288.     }
  289.  
  290.     /* gather list of servers to use, may include local server */
  291.     ret = PINT_cached_config_get_server_list(
  292.         s_op->req->u.create.fs_id,
  293.         s_op->req->u.create.attr.u.meta.dist,
  294.         s_op->req->u.create.num_dfiles_req,
  295.         &s_op->req->u.create.layout,
  296.         &s_op->u.create.io_servers,
  297.         &s_op->u.create.num_io_servers);
  298.     if(ret < 0)
  299.     {
  300.         js_p->error_code = ret;
  301.         return SM_ACTION_COMPLETE;
  302.     }
  303.  
  304.     /* layout may have adjusted number of datafiles */
  305.     s_op->req->u.create.attr.u.meta.dfile_count
  306.         = s_op->u.create.num_io_servers;
  307.     s_op->resp.u.create.datafile_count 
  308.         = s_op->u.create.num_io_servers;
  309.     for(i=0; i<s_op->u.create.num_io_servers; i++)
  310.     {
  311.         gossip_debug(GOSSIP_SERVER_DEBUG, "io_server %d: %s\n", 
  312.             i, s_op->u.create.io_servers[i]); 
  313.     }
  314.  
  315.     s_op->resp.u.create.stuffed = 0;
  316.     js_p->error_code = 0;
  317.     return SM_ACTION_COMPLETE;
  318. }
  319.  
  320. static int create_local_datafiles(
  321.         struct PINT_smcb *smcb, job_status_s *js_p)
  322. {
  323.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  324.     int ret = -1;
  325.     job_id_t tmp_id;
  326.     PVFS_handle_extent_array data_handle_ext_array;
  327.     server_configuration_s *config = get_server_config_struct();
  328.     int i;
  329.     int tmp_index = 0;
  330.  
  331.     if(s_op->resp.u.create.stuffed)
  332.     {
  333.         /* only one datafile, and it is local */
  334.         s_op->u.create.handle_array_local_count = 1;
  335.         s_op->u.create.handle_array_remote_count = 0;
  336.     }
  337.     else
  338.     {
  339.         /* figure out how many datafiles need to be local vs. remote */
  340.         s_op->u.create.handle_array_local_count = 0;
  341.         s_op->u.create.handle_array_remote_count = 0;
  342.         for(i=0; i<s_op->u.create.num_io_servers; i++)
  343.         {
  344.             if(!strcmp(s_op->u.create.io_servers[i], config->host_id))
  345.             {
  346.                 s_op->u.create.handle_array_local_count++;
  347.             }
  348.             else
  349.             {
  350.                 s_op->u.create.handle_array_remote_count++;
  351.                 s_op->u.create.remote_io_servers[tmp_index] = 
  352.                     s_op->u.create.io_servers[i];
  353.                 tmp_index++;
  354.             }
  355.         }
  356.     }
  357.  
  358.     gossip_debug(GOSSIP_SERVER_DEBUG, "creating %d local data files\n", 
  359.         s_op->u.create.handle_array_local_count);
  360.     gossip_debug(GOSSIP_SERVER_DEBUG, "creating %d remote data files\n", 
  361.         s_op->u.create.handle_array_remote_count);
  362.  
  363.     if(s_op->u.create.handle_array_local_count == 0)
  364.     {
  365.         /* no local work to do */
  366.         js_p->error_code = 0;
  367.         return(SM_ACTION_COMPLETE);
  368.     }
  369.  
  370.     /* find local extent array */
  371.     ret = PINT_cached_config_get_server(
  372.         s_op->req->u.create.fs_id,
  373.         config->host_id,
  374.         PINT_SERVER_TYPE_IO,
  375.         &data_handle_ext_array);
  376.     if(ret < 0)
  377.     {
  378.         js_p->error_code = ret;
  379.         return(SM_ACTION_COMPLETE);
  380.     }
  381.  
  382.     /* deliberately not setting SYNC flag, because both the attrs and
  383.      * keyvals will be synced in later states
  384.      */
  385.     ret = job_trove_dspace_create_list(
  386.         s_op->req->u.create.fs_id,
  387.         &data_handle_ext_array,
  388.         s_op->u.create.handle_array_local,
  389.         s_op->u.create.handle_array_local_count,
  390.         PVFS_TYPE_DATAFILE,
  391.         NULL,
  392.         0,
  393.         smcb,
  394.         0,
  395.         js_p,
  396.         &tmp_id,
  397.         server_job_context,
  398.         s_op->req->hints);
  399.  
  400.     return(ret);
  401. }
  402.  
  403. static PINT_sm_action request_datafiles(
  404.     struct PINT_smcb *smcb, job_status_s *js_p)
  405. {
  406.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  407.     int ret = -1;
  408.     job_id_t j_id;
  409.  
  410.     if(s_op->u.create.handle_array_remote_count == 0)
  411.     {
  412.         js_p->error_code = 0;
  413.         return(SM_ACTION_COMPLETE);
  414.     }
  415.  
  416.     ret = job_precreate_pool_get_handles(
  417.         s_op->req->u.create.fs_id,
  418.         s_op->u.create.handle_array_remote_count,
  419.         PVFS_TYPE_DATAFILE,
  420.         s_op->u.create.remote_io_servers,
  421.         s_op->u.create.handle_array_remote,
  422.         0,
  423.         smcb,
  424.         0,
  425.         js_p,
  426.         &j_id,
  427.         server_job_context,
  428.         s_op->req->hints);
  429.     return ret;
  430. }
  431.  
  432. static PINT_sm_action remove_metafile_object(
  433.     struct PINT_smcb *smcb, job_status_s *js_p)
  434. {
  435.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  436.     int ret = -1;
  437.     job_id_t j_id;
  438.  
  439.     /* save the error code before we begin cleanup */
  440.     if(!s_op->u.create.saved_error_code)
  441.     {
  442.         s_op->u.create.saved_error_code = js_p->error_code;
  443.     }
  444.  
  445.     ret = job_trove_dspace_remove(
  446.         s_op->req->u.create.fs_id,
  447.         s_op->resp.u.create.metafile_handle,
  448.         0,
  449.         smcb,
  450.         0,
  451.         js_p,
  452.         &j_id,
  453.         server_job_context,
  454.         s_op->req->hints);
  455.     return ret;
  456. }
  457.  
  458. static PINT_sm_action remove_local_datafile_handles(
  459.     struct PINT_smcb *smcb, job_status_s *js_p)
  460. {
  461.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  462.     int ret = -1;
  463.     job_id_t j_id;
  464.  
  465.     /* save the error code before we begin cleanup */
  466.     if(!s_op->u.create.saved_error_code)
  467.     {
  468.         s_op->u.create.saved_error_code = js_p->error_code;
  469.     }
  470.  
  471.     if(s_op->u.create.handle_array_local_count == 0)
  472.     {
  473.         /* nothing to do */
  474.         js_p->error_code = 0;
  475.         return(SM_ACTION_COMPLETE);
  476.     }
  477.  
  478.     ret = job_trove_dspace_remove_list(s_op->req->u.create.fs_id,
  479.         s_op->u.create.handle_array_local,
  480.         NULL,
  481.         s_op->u.create.handle_array_local_count,
  482.         0,
  483.         smcb,
  484.         0,
  485.         js_p,
  486.         &j_id,
  487.         server_job_context,
  488.         s_op->req->hints);
  489.  
  490.     return ret;
  491. }
  492.  
  493. static PINT_sm_action replace_remote_datafile_handles(
  494.     struct PINT_smcb *smcb, job_status_s *js_p)
  495. {
  496.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  497.     int ret = -1;
  498.     job_id_t j_id;
  499.     PVFS_handle pool_handle;
  500.  
  501.     /* save the error code before we begin cleanup */
  502.     if(!s_op->u.create.saved_error_code)
  503.     {
  504.         s_op->u.create.saved_error_code = js_p->error_code;
  505.     }
  506.  
  507.     if(s_op->u.create.handle_index < s_op->u.create.handle_array_remote_count)
  508.     {
  509.         /* find pool that this handle belongs to */
  510.         ret = job_precreate_pool_lookup_server(
  511.             s_op->u.create.remote_io_servers[s_op->u.create.handle_index],
  512.             PVFS_TYPE_DATAFILE,
  513.             s_op->req->u.create.fs_id,
  514.             &pool_handle);
  515.         if(ret < 0)
  516.         {
  517.             s_op->u.create.handle_index++;
  518.             js_p->error_code = ret;
  519.             return(SM_ACTION_COMPLETE);
  520.         }
  521.  
  522.         /* return handle to pool */
  523.         ret = job_precreate_pool_fill(
  524.             pool_handle,
  525.             s_op->req->u.create.fs_id,
  526.             &s_op->u.create.handle_array_remote[s_op->u.create.handle_index],
  527.             1,
  528.             smcb,
  529.             0,
  530.             js_p,
  531.             &j_id,
  532.             server_job_context,
  533.             s_op->req->hints);
  534.  
  535.         s_op->u.create.handle_index++;
  536.         return(ret);
  537.     }
  538.     else
  539.     {
  540.         /* all handles have been replaced */
  541.         js_p->error_code = REPLACE_DONE;
  542.         return(SM_ACTION_COMPLETE);
  543.     }
  544. }
  545.  
  546. static PINT_sm_action remove_keyvals(
  547.     struct PINT_smcb *smcb, job_status_s *js_p)
  548. {
  549.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  550.     int ret = -1;
  551.     job_id_t j_id;
  552.  
  553.     /* save the error code before we begin cleanup */
  554.     if(!s_op->u.create.saved_error_code)
  555.     {
  556.         s_op->u.create.saved_error_code = js_p->error_code;
  557.     }
  558.  
  559.     /* the keyval keys and vals should still be valid here */
  560.     ret = job_trove_keyval_remove_list(
  561.         s_op->req->u.create.fs_id,
  562.         s_op->resp.u.create.metafile_handle,
  563.         s_op->key_a, s_op->val_a, s_op->error_a,
  564.         2, TROVE_SYNC, NULL, smcb, 0, js_p, &j_id, server_job_context,
  565.         s_op->req->hints);
  566.  
  567.     return ret;
  568. }
  569.  
  570. static PINT_sm_action setup_local_datafile_handles(
  571.     struct PINT_smcb *smcb, job_status_s *js_p)
  572. {
  573.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  574.     int i;
  575.     int tmp_index = 0;
  576.     server_configuration_s *config = get_server_config_struct();
  577.  
  578.     if(s_op->resp.u.create.stuffed)
  579.     {
  580.         s_op->resp.u.create.datafile_handles[0] = 
  581.             s_op->u.create.handle_array_local[0];
  582.         js_p->error_code = 0;
  583.         return(SM_ACTION_COMPLETE);
  584.     }
  585.     else
  586.     {
  587.         for(i=0; i<s_op->u.create.num_io_servers; i++)
  588.         {
  589.             /* find local server positions and set handles */
  590.             if(!strcmp(s_op->u.create.io_servers[i], config->host_id))
  591.             {
  592.                 s_op->resp.u.create.datafile_handles[i] = 
  593.                     s_op->u.create.handle_array_local[tmp_index];
  594.                 tmp_index++;
  595.             }
  596.         }
  597.     }
  598.  
  599.     return SM_ACTION_COMPLETE;
  600. }
  601.  
  602. static PINT_sm_action write_keyvals(
  603.     struct PINT_smcb *smcb, job_status_s *js_p)
  604. {
  605.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  606.     int ret = -1;
  607.     job_id_t j_id;
  608.     int i;
  609.     int tmp_index = 0;
  610.     server_configuration_s *config = get_server_config_struct();
  611.     char* tmpbuf;
  612.  
  613.     if(s_op->u.create.handle_array_remote_count)
  614.     {
  615.         for(i=0; i<s_op->u.create.num_io_servers; i++)
  616.         {
  617.             /* find remote server positions and set handles */
  618.             if(strcmp(s_op->u.create.io_servers[i], config->host_id))
  619.             {
  620.                 s_op->resp.u.create.datafile_handles[i] = 
  621.                     s_op->u.create.handle_array_remote[tmp_index];
  622.                 tmp_index++;
  623.             }
  624.         }
  625.     }
  626.  
  627.     /* start with 2 keyvals: the distribution and the datafile handles */
  628.     int keyval_count = 2;
  629.  
  630.     if(s_op->resp.u.create.stuffed)
  631.     {
  632.         /* also need to set the layout as a keyval */
  633.         keyval_count+= 2;
  634.     }
  635.  
  636.     s_op->key_a = malloc(sizeof(PVFS_ds_keyval) * keyval_count);
  637.     if(!s_op->key_a)
  638.     {
  639.         js_p->error_code = -PVFS_ENOMEM;
  640.         return SM_ACTION_COMPLETE;
  641.     }
  642.  
  643.     s_op->val_a = malloc(sizeof(PVFS_ds_keyval) * keyval_count);
  644.     if(!s_op->val_a)
  645.     {
  646.         free(s_op->key_a);
  647.         js_p->error_code = -PVFS_ENOMEM;
  648.         return SM_ACTION_COMPLETE;
  649.     }
  650.     memset(s_op->val_a, 0, sizeof(PVFS_ds_keyval) * keyval_count);
  651.  
  652.     s_op->key_a[0].buffer = Trove_Common_Keys[METAFILE_HANDLES_KEY].key;
  653.     s_op->key_a[0].buffer_sz = Trove_Common_Keys[METAFILE_HANDLES_KEY].size;
  654.  
  655.     s_op->val_a[0].buffer = s_op->resp.u.create.datafile_handles;
  656.     s_op->val_a[0].buffer_sz =
  657.          s_op->resp.u.create.datafile_count * sizeof(PVFS_handle);
  658.  
  659.     s_op->key_a[1].buffer = Trove_Common_Keys[METAFILE_DIST_KEY].key;
  660.     s_op->key_a[1].buffer_sz = Trove_Common_Keys[METAFILE_DIST_KEY].size;
  661.  
  662.     s_op->val_a[1].buffer_sz =
  663.         s_op->req->u.create.attr.u.meta.dist_size;
  664.     s_op->val_a[1].buffer = malloc(s_op->val_a[1].buffer_sz);
  665.     if(!s_op->val_a[1].buffer)
  666.     {
  667.         js_p->error_code = -PVFS_ENOMEM;
  668.         return SM_ACTION_COMPLETE;
  669.     }
  670.     PINT_dist_encode(s_op->val_a[1].buffer,
  671.                      s_op->req->u.create.attr.u.meta.dist);
  672.  
  673.     if(s_op->resp.u.create.stuffed)
  674.     {
  675.         s_op->key_a[2].buffer = Trove_Common_Keys[METAFILE_LAYOUT_KEY].key;
  676.         s_op->key_a[2].buffer_sz = Trove_Common_Keys[METAFILE_LAYOUT_KEY].size;
  677.  
  678.         s_op->val_a[2].buffer = malloc(PVFS_REQ_LIMIT_LAYOUT);
  679.         if(!s_op->val_a[2].buffer)
  680.         {
  681.             js_p->error_code = -PVFS_ENOMEM;
  682.             return SM_ACTION_COMPLETE;
  683.         }
  684.         tmpbuf = s_op->val_a[2].buffer;
  685.         encode_PVFS_sys_layout(&tmpbuf, &s_op->req->u.create.layout);
  686.  
  687.         s_op->val_a[2].buffer_sz = (tmpbuf - (char*)s_op->val_a[2].buffer);
  688.  
  689.         gossip_debug(GOSSIP_SERVER_DEBUG, 
  690.             "create storing layout of size: %d\n", 
  691.             s_op->val_a[2].buffer_sz);
  692.  
  693.         s_op->key_a[3].buffer = Trove_Common_Keys[NUM_DFILES_REQ_KEY].key;
  694.         s_op->key_a[3].buffer_sz = Trove_Common_Keys[NUM_DFILES_REQ_KEY].size;
  695.  
  696.         gossip_debug(
  697.             GOSSIP_SERVER_DEBUG, "create storing NUM_DFILES_REQ_KEY value of %d\n",
  698.             s_op->req->u.create.num_dfiles_req);
  699.         s_op->val_a[3].buffer = &s_op->req->u.create.num_dfiles_req;
  700.         s_op->val_a[3].buffer_sz = sizeof(s_op->req->u.create.num_dfiles_req);
  701.     }
  702.  
  703.     ret = job_trove_keyval_write_list(
  704.         s_op->req->u.create.fs_id,
  705.         s_op->resp.u.create.metafile_handle,
  706.         s_op->key_a, s_op->val_a,
  707.         keyval_count, TROVE_SYNC, NULL, smcb,
  708.         0, js_p, &j_id, server_job_context,
  709.         s_op->req->hints);
  710.     return ret;
  711. }
  712.  
  713. static PINT_sm_action setattr_setobj_attribs(
  714.         struct PINT_smcb *smcb, job_status_s *js_p)
  715. {
  716.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  717.     int ret = -1;
  718.     job_id_t j_id;
  719.     PVFS_object_attr *a_p = NULL;
  720.     PVFS_object_attr *dspace_a_p = NULL;
  721.     PVFS_ds_attributes *ds_attr = NULL;
  722.  
  723.     dspace_a_p = &s_op->attr;
  724.     a_p = &s_op->req->u.create.attr;
  725.  
  726.      /* 
  727.       * Remember that mtime is versioned on disk! so convert it here..
  728.       * It is better to do it here than change the PVFS_object_attr_overwrite_setable
  729.       * macro, since there are many more users of it, I think.
  730.       */
  731.      if (a_p->mask & PVFS_ATTR_COMMON_MTIME_SET)
  732.      {
  733.          PVFS_time orig_mtime = a_p->mtime;
  734.          a_p->mtime = PINT_util_mktime_version(orig_mtime);
  735.          gossip_debug(GOSSIP_SETATTR_DEBUG, "setting version "
  736.                  "to %llu\n\tmtime is %llu\n",
  737.                  llu(a_p->mtime), llu(orig_mtime));
  738.      }
  739.  
  740.      /*
  741.         we have the attribs stored in the dspace, as well as the
  742.         requested attribs to store.  overwrite the ones that are setable
  743.         and specified by the mask value in the request; macro defined in
  744.         pvfs2-storage.h
  745.         */
  746.      PVFS_object_attr_overwrite_setable(dspace_a_p, a_p);
  747.  
  748.      gossip_debug(
  749.          GOSSIP_SERVER_DEBUG,
  750.          "[STUFFED CREATE]: WRITING attrs: [owner = %d, group = %d\n\t"
  751.          "perms = %o, type = %d, atime = %llu, mtime = %llu\n\t"
  752.          "ctime = %llu | dfile_count = %d | dist_size = %d]\n",
  753.          dspace_a_p->owner, dspace_a_p->group, dspace_a_p->perms,
  754.          dspace_a_p->objtype, llu(dspace_a_p->atime),
  755.          llu(PINT_util_mkversion_time(dspace_a_p->mtime)), llu(dspace_a_p->ctime),
  756.          (int)dspace_a_p->u.meta.dfile_count,
  757.          (int)dspace_a_p->u.meta.dist_size);
  758.  
  759.      /* translate attrs to storage attr format */
  760.      ds_attr = &(s_op->ds_attr);
  761.      PVFS_object_attr_to_ds_attr(dspace_a_p, ds_attr);
  762.  
  763.      ret = job_trove_dspace_setattr(
  764.          s_op->req->u.create.fs_id, s_op->resp.u.create.metafile_handle,
  765.          ds_attr,
  766.          TROVE_SYNC,
  767.          smcb, 0, js_p, &j_id, server_job_context,
  768.          s_op->req->hints);
  769.  
  770.      return ret;
  771. }
  772.  
  773. static int setup_resp(
  774.         struct PINT_smcb *smcb, job_status_s *js_p)
  775. {
  776.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  777.     if (js_p->error_code == 0)
  778.     {
  779.         PINT_ACCESS_DEBUG(s_op, GOSSIP_ACCESS_DEBUG,
  780.                           "create: new metadata handle: %llu.\n",
  781.                           llu(s_op->resp.u.create.metafile_handle));
  782.     }
  783.  
  784.     return SM_ACTION_COMPLETE;
  785. }
  786.  
  787.  
  788. /*
  789.  * Function: create_cleanup
  790.  *
  791.  * Params:   server_op *b, 
  792.  *           job_status_s* js_p
  793.  *
  794.  * Pre:      None
  795.  *
  796.  * Post:     None
  797.  *
  798.  * Returns:  int
  799.  *
  800.  * Synopsis: free memory and return
  801.  *           
  802.  */
  803. static int cleanup(
  804.         struct PINT_smcb *smcb, job_status_s *js_p)
  805. {
  806.     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
  807.  
  808.     if(s_op->key_a)
  809.     {
  810.         free(s_op->key_a);
  811.     }
  812.  
  813.     if(s_op->val_a)
  814.     {
  815.         if(s_op->val_a[1].buffer)
  816.         {
  817.             free(s_op->val_a[1].buffer);
  818.         }
  819.         if(s_op->resp.u.create.stuffed && s_op->val_a[2].buffer)
  820.         {
  821.             free(s_op->val_a[2].buffer);
  822.         }
  823.         free(s_op->val_a);
  824.     }
  825.  
  826.     if(s_op->resp.u.create.datafile_handles)
  827.     {
  828.         free(s_op->resp.u.create.datafile_handles);
  829.     }
  830.  
  831.     if(s_op->u.create.handle_array_remote)
  832.     {
  833.         free(s_op->u.create.handle_array_remote);
  834.     }
  835.  
  836.     if(s_op->u.create.handle_array_local)
  837.     {
  838.         free(s_op->u.create.handle_array_local);
  839.     }
  840.  
  841.     if(s_op->u.create.io_servers)
  842.     {
  843.         free(s_op->u.create.io_servers);
  844.     }
  845.     
  846.     if(s_op->u.create.remote_io_servers)
  847.     {
  848.         free(s_op->u.create.remote_io_servers);
  849.     }
  850.  
  851.     return(server_state_machine_complete(smcb));
  852. }
  853.  
  854. static inline int PINT_get_object_ref_create(
  855.     struct PVFS_server_req *req, PVFS_fs_id *fs_id, PVFS_handle *handle)
  856. {
  857.     *fs_id = req->u.create.fs_id;
  858.     *handle = PVFS_HANDLE_NULL;
  859.     return 0;
  860. };
  861.  
  862. struct PINT_server_req_params pvfs2_create_params =
  863. {
  864.     .string_name = "create",
  865.     .get_object_ref = PINT_get_object_ref_create,
  866.     .perm = PINT_SERVER_CHECK_NONE,
  867.     .access_type = PINT_server_req_modify,
  868.     .state_machine = &pvfs2_create_sm
  869. };
  870.  
  871. /*
  872.  * Local variables:
  873.  *  mode: c
  874.  *  c-indent-level: 4
  875.  *  c-basic-offset: 4
  876.  * End:
  877.  *
  878.  * vim: ft=c ts=8 sts=4 sw=4 expandtab
  879.  */
  880.