Changeset 7827

Timestamp:

06/17/09 12:17:31 (4 years ago)

Author:

sson

Message:

Updated the buffer copy method from the nested state machine to parent.

Location:

branches/as-branch/src/server

Files:

• allreduce.sm (modified) (3 diffs)
• io.sm (modified) (9 diffs)
• pipeline.sm (modified) (15 diffs)
• pvfs2-server.h (modified) (3 diffs)

branches/as-branch/src/server/allreduce.sm

@@ -140 +140 @@
     double *a = s_op->u.allreduce.send_buf;
     double *b = s_op->u.allreduce.recv_buf;
-    double tmp;
-    js_p->error_code = 0;
 
     s_op->u.allreduce.current_depth++;
@@ -147 +145 @@
 
     a[0] = a[0] + b[0];
-    s_op->u.allreduce.send_buf = (void*)a;
+    memcpy(s_op->u.allreduce.send_buf, (void*)a, sizeof(double));
 
     gossip_debug(GOSSIP_IO_DEBUG, "received=%lf\n", *b);
@@ -155 +153 @@
     if(s_op->u.allreduce.current_depth < s_op->u.allreduce.tree_depth)
         js_p->error_code = CONTINUE_ALLREDUCE;
+    else
+        js_p->error_code = 0;
 
-    gossip_debug(GOSSIP_IO_DEBUG, "%s: reduced result=%lf\n", __func__,
-                 *a);
+    gossip_debug(GOSSIP_IO_DEBUG, "%s: reduced result=%lf\n", 
+                 __func__, *a);
 
     return SM_ACTION_COMPLETE;

branches/as-branch/src/server/io.sm

@@ -200 +200 @@
     gossip_debug(GOSSIP_IO_DEBUG, "%s: tag=%d\n", __func__, s_op->tag);
     s_op->u.io.user_ptr = NULL;
-    s_op->u.io.op = s_op->req->u.io.op; /* AS: operation */
-    s_op->u.io.datatype = s_op->req->u.io.datatype; /* AS: dtype */
-    s_op->u.io.dfile_array  = s_op->req->u.io.dfile_array; /* AD: dfile_array */
     
     gossip_debug(GOSSIP_IO_DEBUG, "server_ct=%d\n", s_op->req->u.io.server_ct);
@@ -208 +205 @@
     for(i=0; i<s_op->req->u.io.server_ct; i++) {
         gossip_debug(GOSSIP_IO_DEBUG, "dfile_array[%d]=%llu\n", i, 
-                     llu(s_op->u.io.dfile_array[i]));
-        if(s_op->u.io.dfile_array[i] == s_op->req->u.io.handle) {
+                     llu(s_op->req->u.io.dfile_array[i]));
+        if(s_op->req->u.io.dfile_array[i] == s_op->req->u.io.handle) {
             dfile_index = i;
             gossip_debug(GOSSIP_IO_DEBUG, "dfile_index=%d\n", i);
@@ -260 +257 @@
     if(!fs_config) {
         gossip_err("%s: Failed to get filesystem "
-                   "config from fs_id of: %d\n", __func__,
+d                  "config from fs_id of: %d\n", __func__,
                    s_op->u.pipeline.fs_id);
         js_p->error_code = -PVFS_EINVAL;
@@ -289 +286 @@
         memset(pipeline_op, 0, sizeof(*pipeline_op));
 
+        pipeline_op->u.pipeline.op = s_op->req->u.io.op;
+        pipeline_op->u.pipeline.datatype = s_op->req->u.io.datatype;
+
         pipeline_op->u.pipeline.address = s_op->addr;
         pipeline_op->u.pipeline.handle = s_op->req->u.io.handle;
@@ -295 +295 @@
         pipeline_op->u.pipeline.seg_handle = seg_handle;
         pipeline_op->u.pipeline.id = id[i];
-        //pipeline_op->u.pipeline.parent = s_op;
         pipeline_op->u.pipeline.hints = s_op->req->hints;
         pipeline_op->u.pipeline.tag = s_op->tag;
@@ -304 +303 @@
         pipeline_op->u.pipeline.dfile_count = s_op->u.io.file_data.server_ct;
         pipeline_op->u.pipeline.dist = s_op->u.io.file_data.dist;
+        pipeline_op->u.pipeline.dfile_array = s_op->req->u.io.dfile_array;
+        pipeline_op->u.pipeline.file_data = s_op->u.io.file_data;
+        pipeline_op->u.pipeline.file_req_offset = s_op->u.io.file_req_offset;
 
         /* figure out if the fs config has trove data sync turned on or off
@@ -382 +384 @@
         if(pipeline_op->u.pipeline.buffer) {
             if(s_op->req->u.io.io_type == PVFS_IO_READ) {
+                memcpy(s_op->u.io.tmp_buffer, pipeline_op->u.pipeline.buffer, sizeof(double));
                 BMI_memfree(pipeline_op->u.pipeline.address, 
                             pipeline_op->u.pipeline.buffer, 
@@ -575 +578 @@
             s_op->u.io.total_transferred; /* AS */
 
-    gossip_debug(GOSSIP_IO_DEBUG, "%s: total_transferred=%lld\n", __func__,
-    lld(s_op->u.io.total_transferred)); err = PINT_encode(
+    gossip_debug(GOSSIP_IO_DEBUG, "%s: total_transferred=%lld\n", 
+                 __func__, lld(s_op->u.io.total_transferred)); 
+    err = PINT_encode(
         &s_op->resp, PINT_ENCODE_RESP, &(s_op->encoded),
         s_op->addr, s_op->decoded.enc_type);
 
-    gossip_debug(GOSSIP_IO_DEBUG, "%s: err=%d\n", __func__, err); // err = 0
     if (err < 0)
     {
@@ -597 +600 @@
     gossip_debug(GOSSIP_IO_DEBUG, "%s: err=%d\n", __func__, err); // err = 1
 
-    return err;
+    js_p->error_code = err;
+    return SM_ACTION_COMPLETE;
 }
 

branches/as-branch/src/server/pipeline.sm

@@ -31 +31 @@
 #define DO_COMP 103
 #define DO_ALLREDUCE 104
-#define CONTINUE_ALLREDUCE 105
 
 static int s2s_comp_fn(
@@ -123 +122 @@
     job_id_t tmp_id;
 
-    gossip_debug(GOSSIP_IO_DEBUG, "smcb->base_frame=%d, frame_count=%d\n", smcb->base_frame, smcb->frame_count);
+    gossip_debug(GOSSIP_IO_DEBUG, "smcb->base_frame=%d, frame_count=%d\n", 
+                 smcb->base_frame, smcb->frame_count);
     s_op->u.pipeline.buffer_used = 0;
     bytes = s_op->u.pipeline.buffer_size;
@@ -220 +220 @@
     job_id_t tmp_id;
     struct server_configuration_s *user_opts = get_server_config_struct();
-    struct PINT_server_op *parent_s_op = PINT_sm_frame(smcb->parent_smcb, 0);
 
     if(s_op->u.pipeline.segs == 0) {
@@ -240 +239 @@
         assert(s_op->u.pipeline.buffer_used);
 
-        if(parent_s_op->u.io.op != 0) { /* AS: when op is specified */
+        if(s_op->u.pipeline.op != 0) { /* AS: when op is specified */
             ret = DO_COMP; /* AS: skip sending if op is specified */ 
             gossip_debug(GOSSIP_IO_DEBUG, "%s: parent->op != 0\n", __func__);
@@ -309 +308 @@
 {
     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
-    struct PINT_server_op *parent_s_op = PINT_sm_frame(smcb->parent_smcb, 0);
     js_p->error_code = 0;
-    PVFS_size file_req_offset = parent_s_op->u.io.file_req_offset;
-    PINT_request_file_data fdata = parent_s_op->u.io.file_data;
+    PVFS_size file_req_offset = s_op->u.pipeline.file_req_offset;
+    PINT_request_file_data fdata = s_op->u.pipeline.file_data;
     PVFS_simple_stripe_params *dparam = 
         (PVFS_simple_stripe_params*)fdata.dist->params;
@@ -323 +321 @@
     gossip_debug(GOSSIP_IO_DEBUG, "loff=%lld, file_req_offset=%lld\n", lld(loff), lld(file_req_offset));
 
-    switch(parent_s_op->u.io.datatype) {
+    switch(s_op->u.pipeline.datatype) {
     case ((int)0x4c000405): /* MPI_INT */
         count = (PVFS_size)(s_op->u.pipeline.buffer_used - 
@@ -365 +363 @@
 {
     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
-    struct PINT_server_op *parent_s_op = PINT_sm_frame(smcb->parent_smcb, 0);
     PINT_sm_msgpair_state *msg_p = NULL;
     struct server_configuration_s *user_opts = get_server_config_struct();
@@ -388 +385 @@
     /* determine which server we need to talk to */
     next_server_index = (s_op->u.pipeline.dfile_index + 1)%(s_op->u.pipeline.dfile_count);
-    next_server_handle = parent_s_op->u.io.dfile_array[next_server_index];
+    next_server_handle = s_op->u.pipeline.dfile_array[next_server_index];
 
     /* build a request */
@@ -511 +508 @@
                      "%s: buffer_used=%lld, op=0x%x, datatype=0x%x, actual_size=%lld\n", 
                      __func__, lld(s_op->u.pipeline.buffer_used), 
-                     parent_s_op->u.io.op, 
-                     parent_s_op->u.io.datatype,
+                     s_op->u.pipeline.op, 
+                     s_op->u.pipeline.datatype,
                      lld(js_p->actual_size)); /* AS */
 
-        switch(parent_s_op->u.io.datatype) {
+        switch(s_op->u.pipeline.datatype) {
         case ((int)0x4c000405): /* MPI_INT */
             {
@@ -541 +538 @@
                 gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "count=%lld, tmp=%d\n", lld(count), *tmp);
 
-                switch(parent_s_op->u.io.op) {
+                switch(s_op->u.pipeline.op) {
                 case 0x58000001: /* MAX */
                     result = *a;
@@ -653 +650 @@
                 gossip_debug(GOSSIP_FLOW_PROTO_DEBUG, "count=%lld, tmp=%lf\n", 
                              lld(count), *tmp);
-                switch(parent_s_op->u.io.op) {
+                switch(s_op->u.pipeline.op) {
                 case 0x58000001: /* MAX */
                     result = *a;
@@ -753 +750 @@
 {
     struct PINT_server_op *s_op = PINT_sm_frame(smcb, PINT_FRAME_CURRENT);
-    struct PINT_server_op *parent_s_op = PINT_sm_frame(smcb->parent_smcb, 0);
     int ret, tmp_id;
     struct server_configuration_s *user_opts = get_server_config_struct();
@@ -767 +763 @@
     allreduce_op->u.allreduce.recv_buf = (void*)malloc(5*sizeof(double));
     memset(allreduce_op->u.allreduce.recv_buf, 0, 5*sizeof(double));
-    allreduce_op->u.allreduce.tree_depth = log2(s_op->u.pipeline.dfile_count); /* FIXME !!!! */
+    allreduce_op->u.allreduce.tree_depth = log2(s_op->u.pipeline.dfile_count);
     gossip_debug(GOSSIP_IO_DEBUG, "tree_depth=%d\n", 
                  allreduce_op->u.allreduce.tree_depth);
     allreduce_op->u.allreduce.current_depth = 0;
     allreduce_op->u.allreduce.mask = 0x1;
-    allreduce_op->u.allreduce.dfile_array = parent_s_op->u.io.dfile_array;
+    allreduce_op->u.allreduce.dfile_array = s_op->u.pipeline.dfile_array;
     allreduce_op->u.allreduce.send_buf = (void*)malloc(5*sizeof(double));
     memset(allreduce_op->u.allreduce.send_buf, 0, 5*sizeof(double));
@@ -824 +820 @@
     /* FIXME: */
     /* unless the second condition is set, the server starts
-       a new read request to one already done, and falls into
+       a new read request that is already done, and falls into
        infinite trove_read->bmi_send->check_done loop */
     if(!segpool_done(h) && s_op->u.pipeline.segs != 0) {
@@ -831 +827 @@
     }
     else {
-        if(parent_s_op->u.io.op == (0x5800000f)) { /* KMEANS */
-            parent_s_op->u.io.tmp_buffer = (void *)s_op->u.pipeline.buffer;
+#if 0
+        if(s_op->u.pipeline.op == (0x5800000f)) { /* KMEANS */
+            memcpy(parent_s_op->u.io.tmp_buffer, 
+                   s_op->u.pipeline.buffer, sizeof(double));
         }
+#endif
         gossip_debug(GOSSIP_IO_DEBUG, "%s: DONE\n", __func__);
     }

branches/as-branch/src/server/pvfs2-server.h

@@ -342 +342 @@
     void *user_ptr;
 
-    int op;
-    int datatype;
     void *tmp_buffer;
     PVFS_size count; /* for MEAN operation */
-    PVFS_handle *dfile_array;
 
     PVFS_size aggregate_size;
@@ -367 +364 @@
     PVFS_BMI_addr_t address;
 
+    PVFS_handle *dfile_array;
     int dfile_index; /* can be used for Rank */
     int dfile_count;
     struct PINT_dist_s *dist;
+    PINT_request_file_data file_data;
 
     PINT_Request *file_req;
@@ -375 +374 @@
     PINT_Request *mem_req;
 
+    /* for strip alignment */
     char tmp_buf[128]; /* FIXME */
     PVFS_size unaligned_size;
 
     enum PVFS_io_type io_type;
+
+    /* AS: operator and data type */
+    int op;
+    int datatype;
 
     char *buffer;