working on bulk transfer all

Author: TheAssembler1

run_client.sh

@@ -4,6 +4,7 @@
 
 pushd ./build/bin || exit 1  # Exit if cd fails
 export PDC_DEBUG=1
-"./region_transfer_2D_partial"
+#"./region_transfer_all_append"
+"./sandbox"
 popd
 

run_test.sh

@@ -11,5 +11,6 @@ echo "deleting existing data"
 rm -rf /home/nlewi26/src/work_space/source/pdc/build/bin/pdc_data
 
 pushd ./build
-ctest -L serial
+#ctest -L serial --stop-on-failure --output-on-failure
+ctest -L serial -I 1,37 -I 39,68
 popd

src/server/include/pdc_client_server_common.h

@@ -338,7 +338,7 @@ typedef struct data_server_region_t {
     // Used primarily as a local_temp
     void *obj_data_ptr;
     // FIXME: (Noah) we should dynamically allocate this number....
-    char *                       storage_location[1024]; // save the file location to enable reopening
+    char *                       storage_location[4096]; // save the file location to enable reopening
     // number of elements per region
     int region_size_elements;
     struct data_server_region_t *prev;

src/server/pdc_server_region/pdc_server_data.c

@@ -37,6 +37,7 @@
 #include <math.h>
 #include <sys/shm.h>
 #include <sys/mman.h>
+#include <errno.h>
 
 #include "pdc_config.h"
 
@@ -126,7 +127,16 @@ static int
 server_open_storage(char *storage_location, pdcid_t obj_id, uint64_t region_index)
 {
     fill_storage_path(storage_location, obj_id, region_index);
-    return open(storage_location, O_RDWR | O_CREAT, 0666);
+
+    errno = 0;
+    int fd = open(storage_location, O_RDWR | O_CREAT, 0666);
+
+    if(fd <= 0) {
+        LOG_ERROR("Failed to open file: %s\n", storage_location);
+        perror("open");
+    } 
+
+    return fd;
 }
 
 perr_t
@@ -395,6 +405,7 @@ PDC_Server_register_obj_region_by_pointer(data_server_region_t **new_obj_reg_ptr
     else {
         if (new_obj_reg->fd == -1) {
             // FIXME: (Noah)
+            new_obj_reg->storage_location[region_index] = (char *)malloc(sizeof(char) * ADDR_MAX);
             fill_storage_path(new_obj_reg->storage_location[region_index], obj_id, region_index);
         }
         if (new_obj_reg->fd < 0) {
@@ -406,6 +417,8 @@ PDC_Server_register_obj_region_by_pointer(data_server_region_t **new_obj_reg_ptr
             // FIXME: (Noah)
             new_obj_reg->fd = open(new_obj_reg->storage_location[region_index], O_RDWR | O_CREAT, 0666);
             if (new_obj_reg->fd < 0) {
+                LOG_ERROR("Failed to open file %s\n", new_obj_reg->storage_location[region_index]);
+                perror("open");
                 goto done;
             }
         }
@@ -4536,9 +4549,9 @@ PDC_Server_data_io_from_region_per_file(uint64_t obj_id, struct pdc_region_info
     uint64_t *offset = region_info->offset;
     int ndim = region_info->ndim;
 
-    printf("\tobj meta ndim: %d\n", ndim);
+    LOG_JUST_PRINT("\tobj meta ndim: %d\n", ndim);
     for (int i = 0; i < ndim; i++) {
-        printf("\tobj meta dims[%d] = %" PRIu64 "\n", i, global_size[i]);
+        LOG_JUST_PRINT("\tobj meta dims[%d] = %" PRIu64 "\n", i, global_size[i]);
     }
 
     // Compute flat offset from global start
@@ -4547,25 +4560,38 @@ PDC_Server_data_io_from_region_per_file(uint64_t obj_id, struct pdc_region_info
         flat_offset += offset[i] * stride;
         stride *= global_size[i];
     }
+    
+    LOG_JUST_PRINT("\tnum_elements: %d\n", num_elements);
+    LOG_JUST_PRINT("\tflat_offset: %d\n", flat_offset);
 
     // Determine initial region index
     uint64_t region_index = 0;
     if (region == NULL) {
-        region_index = 0;
+        region_index = flat_offset / num_elements;
     } else {
+        if(region->region_size_elements == UNITIALIZED_DATA_SERVER_REGION_SIZE_ELEMENTS) {
+            region->region_size_elements = num_elements;
+        }
         region_index = flat_offset / region->region_size_elements;
     }
 
+    LOG_JUST_PRINT("\tregion_index: %d\n", region_index);
+
     // Register the initial region
     PDC_Server_register_obj_region_by_pointer(&region, obj_id, 0, region_index);
 
+    if(region->region_size_elements == UNITIALIZED_DATA_SERVER_REGION_SIZE_ELEMENTS) {
+        region->region_size_elements = num_elements;
+    }
+
     // Allocate indices for iteration
     uint64_t *indices = (uint64_t *)calloc(ndim, sizeof(uint64_t));
     if (indices == NULL) {
         PGOTO_ERROR(FAIL, "calloc failed for indices");
     }
 
     for (uint64_t count = 0; count < num_elements; count++) {
+        printf("cur count: %d\n", count);
         // Compute flat index
         uint64_t flat_index = 0;
         uint64_t temp_stride = 1;
@@ -4577,33 +4603,40 @@ PDC_Server_data_io_from_region_per_file(uint64_t obj_id, struct pdc_region_info
         // Determine region
         uint64_t new_region_index = flat_index / region->region_size_elements;
         if (new_region_index != region_index) {
+            uint64_t old_region_size = region->region_size_elements;
             if (region) {
                 PDC_Server_unregister_obj_region_by_pointer(region, 0);
                 region = NULL;
             }
             region_index = new_region_index;
             PDC_Server_register_obj_region_by_pointer(&region, obj_id, 0, region_index);
+            region->region_size_elements = old_region_size;
         }
 
         if (region == NULL || region->region_size_elements == 0) {
             PGOTO_ERROR(FAIL, "Region not properly initialized");
         }
 
-        // Compute offset within region
         uint64_t offset_region = flat_index % region->region_size_elements;
         void *target_buf = (char *)buf + (count * unit);
         off_t file_offset = offset_region * unit;
 
         ssize_t res;
+        errno = 0;
         if (io_type == REGION_PER_FILE_IO_TYPE_READ) {
             res = pread(region->fd, target_buf, unit, file_offset);
         } else {
             res = pwrite(region->fd, target_buf, unit, file_offset);
         }
 
         if (res != (ssize_t)unit) {
+            LOG_JUST_PRINT("expected io size %d: actual io size %d\n", unit, res);
+            LOG_JUST_PRINT("region size %d, region index: %d\n", region->region_size_elements, region_index);
+            LOG_JUST_PRINT("io error storage location: %s\n", region->storage_location[region_index]);
             perror(io_type == REGION_PER_FILE_IO_TYPE_READ ? "pread" : "pwrite");
-            PGOTO_ERROR(FAIL, "I/O operation failed");
+            PGOTO_ERROR(FAIL, "I/O operation failed\n");
+        } else {
+            LOG_JUST_PRINT("io storage location: %s\n", region->storage_location[region_index]);
         }
 
         // Advance indices

src/server/pdc_server_region/pdc_server_region_request_handler.h

@@ -79,7 +79,7 @@ transfer_request_all_bulk_transfer_read_cb(const struct hg_cb_info *info)
     for (i = 0; i < request_data.n_objs; ++i) {
         temp_ptrs[i] = PDC_Server_get_obj_region(request_data.obj_id[i]);
         // FIXME: Noah
-        if(temp_ptrs[i] == NULL) {
+        if(temp_ptrs[i]->region_size_elements == UNITIALIZED_DATA_SERVER_REGION_SIZE_ELEMENTS) {
             LOG_ERROR("GOING TO GENERATE INCORRECT REGION INDEXES\n");
         }
         PDC_Server_register_obj_region_by_pointer(&(temp_ptrs[i]), request_data.obj_id[i], 1, 0);
@@ -174,11 +174,6 @@ transfer_request_all_bulk_transfer_write_cb(const struct hg_cb_info *info)
 
     FUNC_ENTER(NULL);
 
-#ifdef TANG_DEBUG
-    PDC_get_time_str(cur_time);
-    LOG_DEBUG("%s ==PDC_SERVER[%d]: enter\n", cur_time, PDC_get_rank());
-#endif
-
     gettimeofday(&last_cache_activity_timeval_g, NULL);
 
 #ifdef PDC_TIMING
@@ -197,17 +192,36 @@ transfer_request_all_bulk_transfer_write_cb(const struct hg_cb_info *info)
 #ifndef PDC_SERVER_CACHE
     data_server_region_t **temp_ptrs =
         (data_server_region_t **)malloc(sizeof(data_server_region_t *) * request_data.n_objs);
+
+    LOG_JUST_PRINT("========================REQUEST ALL INFO===============================\n");
     for (i = 0; i < request_data.n_objs; ++i) {
+        LOG_JUST_PRINT("obj_id: %d\n", request_data.obj_id[i]);
+        LOG_JUST_PRINT("\tobj ndim: %d\n", request_data.obj_ndim[i]);
+        LOG_JUST_PRINT("\tobj unit: %d\n", request_data.unit[i]);
+        LOG_JUST_PRINT("\tregion ndim[%d]=%d\n", i, request_data.remote_ndim[i]);
+        for(int j = 0; j < request_data.remote_ndim[i]; j++) {
+            LOG_JUST_PRINT("\tregion size[%d]=%d\n", j, request_data.remote_length[i][j]);
+            LOG_JUST_PRINT("\tregion offset[%d]=%d\n", j, request_data.remote_offset[i][j]);
+        }
+
         temp_ptrs[i] = PDC_Server_get_obj_region(request_data.obj_id[i]);
+
         // FIXME: Noah
-        LOG_ERROR("\n\n\n\n NOT IMPLEMENTED \n\n\n\n\n");
-        PDC_Server_register_obj_region_by_pointer(temp_ptrs + i, request_data.obj_id[i], 1, 0);
-    }
-#endif
+        // Compute total number of elements
+        uint64_t num_elements = 1;
+        // FIXME: this loops through the obj_ndim instead of the remote ndim
+        for (int j = 0; j < request_data.obj_ndim[i]; j++) {
+            num_elements *= request_data.remote_length[i][j];
+        }
 
-#ifdef TANG_DEBUG
-    PDC_get_time_str(cur_time);
-    LOG_DEBUG("%s ==PDC_SERVER[%d]: before (cache) writing\n", cur_time, PDC_get_rank());
+        if(temp_ptrs[i]->region_size_elements == UNITIALIZED_DATA_SERVER_REGION_SIZE_ELEMENTS) {
+            temp_ptrs[i]->region_size_elements = num_elements;
+        }
+
+        uint64_t region_index = num_elements / temp_ptrs[i]->region_size_elements;
+        PDC_Server_register_obj_region_by_pointer(temp_ptrs + i, request_data.obj_id[i], 1, region_index);
+    }
+    LOG_JUST_PRINT("=====================================END===============================\n");
 #endif
 
     for (i = 0; i < request_data.n_objs; ++i) {

src/server/pdc_server_region/pdc_server_region_transfer.c

@@ -261,6 +261,7 @@ PDC_transfer_request_id_register()
 static void
 PDC_Server_io_not_by_region_region_per_file(char *user_specified_path, int is_write)
 {
+    LOG_ERROR("\n\n\n\nUNIMPLEMENTED\n\n\n\n");
 }
 
 perr_t

src/tests/region_transfer_2D_partial.c

@@ -182,7 +182,6 @@ main(int argc, char **argv)
     memset(data_read, 0, BUF_LEN);
 
     LOG_INFO("Sleeping for 1 s\n");
-    sleep(2);
 
     transfer_request = PDCregion_transfer_create(data_read, PDC_READ, obj1, reg, reg_global);
 

src/tests/region_transfer_all_append.c

@@ -31,9 +31,9 @@
 #include <unistd.h>
 #include <sys/time.h>
 #include "pdc.h"
-#define BUF_LEN  64
-#define OBJ_NUM  17
-#define REQ_SIZE 8
+#define BUF_LEN  2
+#define OBJ_NUM  1
+#define REQ_SIZE 2
 
 int
 main(int argc, char **argv)
@@ -785,7 +785,7 @@ main(int argc, char **argv)
         if (ret_value == 0)
             LOG_INFO("Test succeed!\n");
         else
-            LOG_ERROR("ErROR: Test failed!\n");
+            LOG_ERROR("Test failed!\n");
     }
 
 #ifdef ENABLE_MPI