all tests pass

Author: TheAssembler1

src/server/pdc_server_region/pdc_server_data.c

@@ -4491,247 +4491,159 @@ _setup_zfp(struct pdc_region_info *region_info, zfp_stream **zfp)
 }
 #endif
 
-#define PRINT_DATA
+typedef enum {
+    IO_READ,
+    IO_WRITE
+} IO_TYPE;
 
-perr_t
-PDC_Server_data_write_out_region_per_file(uint64_t obj_id, struct pdc_region_info *region_info, void *buf,
-                                          size_t unit)
+static perr_t
+PDC_Server_data_io_from_region_per_file(uint64_t obj_id, struct pdc_region_info *region_info, void *buf,
+                                        size_t unit, IO_TYPE io_type)
 {
     int ret_value = SUCCEED;
 
     FUNC_ENTER(NULL);
 
     LOG_JUST_PRINT("===============================================\n");
-    LOG_JUST_PRINT("Called with parameters:\n");
+    LOG_JUST_PRINT("IO (%s) called with parameters:\n", io_type == IO_READ ? "READ" : "WRITE");
     LOG_JUST_PRINT("\tobj_id: %" PRIu64 "\n", obj_id);
     LOG_JUST_PRINT("\tregion_info: %p\n", (void *)region_info);
     LOG_JUST_PRINT("\tbuf: %p\n", buf);
     LOG_JUST_PRINT("\tunit: %zu\n", unit);
 
-    for(int i = 0; i < region_info->ndim; i++) {
-        printf("\tregion_info->offset[%d] = %d\n", i, region_info->offset[i]);
-    }
-    for(int i = 0; i < region_info->ndim; i++) {
-        printf("\tregion_info->size[%d] = %d\n", i, region_info->size[i]);
+    for (int i = 0; i < region_info->ndim; i++) {
+        printf("\tregion_info->offset[%d] = %" PRIu64 "\n", i, region_info->offset[i]);
+        printf("\tregion_info->size[%d]   = %" PRIu64 "\n", i, region_info->size[i]);
     }
 
-    // compute total write size in bytes/elements
-    uint64_t write_size_elements = region_info->size[0];
-    for (int i = 1; i < region_info->ndim; i++) {
-        write_size_elements *= region_info->size[i];
+    // Compute total number of elements
+    uint64_t num_elements = 1;
+    for (int i = 0; i < region_info->ndim; i++) {
+        num_elements *= region_info->size[i];
     }
-    uint64_t write_size_bytes = write_size_elements * unit;
 
-    if (write_size_bytes == 0) {
-        PGOTO_ERROR(FAIL, "Write size was 0");
+    if (num_elements == 0 || unit == 0) {
+        PGOTO_ERROR(FAIL, "Requested I/O with 0 size");
     }
 
-    // compute total write offset in bytes/elements
-    uint64_t write_offset_size_elements = 0; 
-    uint64_t stride = 1;
-    for (int i = region_info->ndim - 1; i >= 0; i--) {
-        write_offset_size_elements += region_info->offset[i] * stride;
-        stride *= region_info->size[i];
+    pdc_metadata_t *obj_meta = PDC_Server_get_obj_metadata(obj_id);
+    if (obj_meta == NULL) {
+        PGOTO_ERROR(FAIL, "Object metadata not found");
     }
 
     data_server_region_t *region = PDC_Server_get_obj_region(obj_id);
+    uint64_t *global_size = obj_meta->dims;
+    uint64_t *offset = region_info->offset;
+    int ndim = region_info->ndim;
 
-    // compute region index
-    uint64_t region_index;
-    if (region == NULL) {
-        region_index = write_offset_size_elements / write_size_elements;
+    printf("\tobj meta ndim: %d\n", ndim);
+    for (int i = 0; i < ndim; i++) {
+        printf("\tobj meta dims[%d] = %" PRIu64 "\n", i, global_size[i]);
     }
-    else {
-        region_index = write_offset_size_elements / region->region_size_elements;
+
+    // Compute flat offset from global start
+    uint64_t flat_offset = 0, stride = 1;
+    for (int i = ndim - 1; i >= 0; i--) {
+        flat_offset += offset[i] * stride;
+        stride *= global_size[i];
     }
 
-    // loop through file regions and update/create if necessary
-    while (write_size_elements > 0) {
-        LOG_JUST_PRINT("\tregion index: %lu\n", region_index);
+    // Determine initial region index
+    uint64_t region_index = 0;
+    if (region == NULL) {
+        region_index = 0;
+    } else {
+        region_index = flat_offset / region->region_size_elements;
+    }
 
-        PDC_Server_register_obj_region_by_pointer(&region, obj_id, 0, region_index);
+    // Register the initial region
+    PDC_Server_register_obj_region_by_pointer(&region, obj_id, 0, region_index);
 
-        if (!region) {
-            LOG_ERROR("Failed to find region with obj_id: %" PRIu64 "\n", obj_id);
-            PGOTO_ERROR(FAIL, "Failed to find region");
-        }
-        if(region->storage_location[region_index] == NULL) {
-            LOG_ERROR("Storage location was NULL\n");
-            PGOTO_ERROR(FAIL, "Storage location was null");
+    // 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++) {
+        // Compute flat index
+        uint64_t flat_index = 0;
+        uint64_t temp_stride = 1;
+        for (int i = ndim - 1; i >= 0; i--) {
+            flat_index += (offset[i] + indices[i]) * temp_stride;
+            temp_stride *= global_size[i];
         }
 
-        // update region size if it has not been initialized
-        if (region->region_size_elements == UNITIALIZED_DATA_SERVER_REGION_SIZE_ELEMENTS) {
-            region->region_size_elements = write_size_elements;
+        // Determine region
+        uint64_t new_region_index = flat_index / region->region_size_elements;
+        if (new_region_index != region_index) {
+            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);
         }
 
-        // get offset remainder of current region write
-        uint64_t remainder_elements = write_offset_size_elements % region->region_size_elements;
+        if (region == NULL || region->region_size_elements == 0) {
+            PGOTO_ERROR(FAIL, "Region not properly initialized");
+        }
 
-        // compute current region write size
-        uint64_t cur_write_size_elements =
-            PDC_MIN(region->region_size_elements - remainder_elements, write_size_elements);
+        // 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;
 
-        if (remainder_elements != 0) {
-            LOG_JUST_PRINT("Shifted by %d\n", remainder_elements * unit);
-            lseek(region->fd, remainder_elements * unit, SEEK_SET);
+        ssize_t res;
+        if (io_type == IO_READ) {
+            res = pread(region->fd, target_buf, unit, file_offset);
         } else {
-            LOG_JUST_PRINT("No shift\n");
-            lseek(region->fd, 0, SEEK_SET);
+            res = pwrite(region->fd, target_buf, unit, file_offset);
         }
 
-        LOG_JUST_PRINT("Write information:\n");
-        LOG_JUST_PRINT("\tstorage location: %s\n", region->storage_location[region_index]);
-        LOG_JUST_PRINT("\tsize bytes: %lu\n", cur_write_size_elements * unit);
-        LOG_JUST_PRINT("\toffset bytes: %lu\n", remainder_elements * unit);
-        LOG_JUST_PRINT("\tfd: %d\n", region->fd);
-
-#ifdef PRINT_DATA
-        for (int i = 0; i < cur_write_size_elements; i++) {
-            int* data = (int*)buf;
-            LOG_JUST_PRINT("[%3d]=%12d  ", i, data[i]);
-            if ((i + 1) % 5 == 0)
-                LOG_JUST_PRINT("\n");
+        if (res != (ssize_t)unit) {
+            perror(io_type == IO_READ ? "pread" : "pwrite");
+            PGOTO_ERROR(FAIL, "I/O operation failed");
         }
-        LOG_JUST_PRINT("\n");
-#endif
 
-        perr_t write_stat = PDC_Server_posix_write(region->fd, buf, cur_write_size_elements * unit);
-        if (write_stat != SUCCEED) {
-            PGOTO_ERROR(FAIL, "Failed to posix write region\n");
+        // Advance indices
+        for (int d = ndim - 1; d >= 0; d--) {
+            indices[d]++;
+            if (indices[d] < region_info->size[d]) {
+                break;
+            } else {
+                indices[d] = 0;
+            }
         }
+    }
 
+done:
+    if (region) {
         PDC_Server_unregister_obj_region_by_pointer(region, 0);
-
-        // Advance state for next region (if needed)
-        buf = (char *)buf + cur_write_size_elements * unit;
-        write_size_elements -= cur_write_size_elements;
-        write_offset_size_elements += cur_write_size_elements;
-        region_index = write_offset_size_elements / region->region_size_elements;
+        region = NULL;
     }
 
+    if (indices)
+        free(indices);
+
     LOG_JUST_PRINT("===============================================\n");
 
-done:
     FUNC_LEAVE(ret_value);
 }
 
+
 perr_t
-PDC_Server_data_read_from_region_per_file(uint64_t obj_id, struct pdc_region_info *region_info, void *buf,
-                                        size_t unit)
+PDC_Server_data_write_out_region_per_file(uint64_t obj_id, struct pdc_region_info *region_info, void *buf,
+                                          size_t unit)
 {
-    int ret_value = SUCCEED;
-
-    FUNC_ENTER(NULL);
-
-    LOG_JUST_PRINT("===============================================\n");
-    LOG_JUST_PRINT("READ called with parameters:\n");
-    LOG_JUST_PRINT("\tobj_id: %" PRIu64 "\n", obj_id);
-    LOG_JUST_PRINT("\tregion_info: %p\n", (void *)region_info);
-    LOG_JUST_PRINT("\tbuf: %p\n", buf);
-    LOG_JUST_PRINT("\tunit: %zu\n", unit);
-
-    for(int i = 0; i < region_info->ndim; i++) {
-        printf("\tregion_info->offset[%d] = %d\n", i, region_info->offset[i]);
-    }
-    for(int i = 0; i < region_info->ndim; i++) {
-        printf("\tregion_info->size[%d] = %d\n", i, region_info->size[i]);
-    }
-
-    // compute total read size in bytes/elements
-    uint64_t read_size_elements = region_info->size[0];
-    for (int i = 1; i < region_info->ndim; i++) {
-        read_size_elements *= region_info->size[i];
-    }
-    uint64_t read_size_bytes = read_size_elements * unit;
-
-    if (read_size_bytes == 0) {
-        PGOTO_ERROR(FAIL, "Read size was 0");
-    }
-
-    // compute total read offset size bytes/elements
-    uint64_t read_offset_elements = 0; 
-    uint64_t stride = 1;
-    for (int i = 0; i < region_info->ndim; i++) {
-        read_offset_elements += region_info->offset[i] * stride;
-        stride *= region_info->size[i];
-    }
-
-    data_server_region_t *region = PDC_Server_get_obj_region(obj_id);
-
-    // compute region index
-    uint64_t region_index;
-    if (region == NULL) {
-        region_index = read_offset_elements / read_size_elements;
-    } else {
-        region_index = read_offset_elements / region->region_size_elements;
-    }
-
-    // loop through file regions and update/create if necessary
-    while (read_size_elements > 0) {
-        LOG_JUST_PRINT("\tregion index: %lu\n", region_index);
-
-        PDC_Server_register_obj_region_by_pointer(&region, obj_id, 0, region_index);
-        if (!region) {
-            LOG_ERROR("Failed to find region with obj_id: %" PRIu64 "\n", obj_id);
-            PGOTO_ERROR(FAIL, "Failed to find region");
-        }
-        if(region->storage_location[region_index] == NULL) {
-            LOG_ERROR("Storage location was NULL\n");
-            PGOTO_ERROR(FAIL, "Storage location was null");
-        }
-
-        // get current region offset
-        uint64_t remainder_elements = read_offset_elements % region->region_size_elements;
-
-        // calculate how much to read from this region
-        uint64_t cur_read_size_elements =
-            PDC_MIN(region->region_size_elements - remainder_elements, read_size_elements);
-
-        // seek to correct offset
-        if (remainder_elements != 0) {
-            LOG_JUST_PRINT("Shifted by %d\n", remainder_elements * unit);
-            lseek(region->fd, remainder_elements * unit, SEEK_SET);
-        } else {
-            LOG_JUST_PRINT("No shift\n");
-            lseek(region->fd, 0, SEEK_SET);
-        }
-
-        LOG_JUST_PRINT("Read information:\n");
-        LOG_JUST_PRINT("\tstorage location: %s\n", region->storage_location[region_index]);
-        LOG_JUST_PRINT("\tsize bytes: %lu\n", cur_read_size_elements * unit);
-        LOG_JUST_PRINT("\toffset bytes: %lu\n", remainder_elements * unit);
-        LOG_JUST_PRINT("\tfd: %d\n", region->fd);
-
-        // read from file
-        ssize_t bytes_read = read(region->fd, buf, cur_read_size_elements * unit);
-        if (bytes_read < 0 || (size_t)bytes_read != cur_read_size_elements * unit) {
-            PGOTO_ERROR(FAIL, "Failed to read from region file");
-        }
-
-#ifdef PRINT_DATA
-        for (int i = 0; i < cur_read_size_elements; i++) {
-            int* data = (int*)buf;
-            LOG_JUST_PRINT("[%3d]=%12d  ", i, data[i]);  
-            if ((i + 1) % 5 == 0)
-                LOG_JUST_PRINT("\n");
-        }
-        LOG_JUST_PRINT("\n");
-#endif
-
-        PDC_Server_unregister_obj_region_by_pointer(region, 0);
-
-        // advance
-        buf = (char *)buf + cur_read_size_elements * unit;
-        read_offset_elements += cur_read_size_elements;
-        read_size_elements -= cur_read_size_elements;
-        region_index = read_offset_elements / region->region_size_elements;
-    }
-
-    LOG_JUST_PRINT("===============================================\n");
+    return PDC_Server_data_io_from_region_per_file(obj_id, region_info, buf, unit, IO_WRITE);
+}
 
-done:
-    FUNC_LEAVE(ret_value);
+perr_t
+PDC_Server_data_read_from_region_per_file(uint64_t obj_id,  struct pdc_region_info *region_info, void *buf,
+                                        size_t unit)
+{
+    return PDC_Server_data_io_from_region_per_file(obj_id, region_info, buf, unit, IO_READ);
 }
 
 perr_t
@@ -4760,7 +4672,6 @@ PDC_Server_data_write_out(uint64_t obj_id, struct pdc_region_info *region_info,
         write_size = unit * region_info->size[0];
     if (region_info->ndim >= 2)
         write_size *= region_info->size[1];
-
     if (region_info->ndim >= 3)
         write_size *= region_info->size[2];
     region = PDC_Server_get_obj_region(obj_id);

src/tests/region_transfer_2D_partial.c

@@ -181,6 +181,9 @@ 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);
 
     PDCregion_transfer_start(transfer_request);
@@ -191,10 +194,8 @@ main(int argc, char **argv)
     for (i = 0; i < BUF_LEN / 2; ++i) {
         if (data_read[i] != i * 2 + 1) {
             LOG_JUST_PRINT("wrong value %d!=%d\n", data_read[i], i * 2 + 1);
-            //ret_value = 1;
-            //break;
-        } else {
-            LOG_JUST_PRINT("right value %d!=%d\n", data_read[i], i * 2 + 1);
+            ret_value = 1;
+            break;
         }
     }
 

src/tests/sandbox.c

@@ -28,7 +28,7 @@
 #include <unistd.h>
 #include "pdc.h"
 
-#define NPARTICLES   100
+#define NPARTICLES   10
 #define PARTICLE_VAL 5.5f
 
 #define CHECK_AND_PRINT_GT(condition, success_msg, fail_msg)                                                 \