Finalize Read Ahead

Author: Omar Marzouk

tensorflow_io/core/filesystems/dfs/dfs_filesystem.cc

@@ -14,13 +14,31 @@ typedef struct DFSRandomAccessFile {
   std::vector<ReadBuffer> buffers;
   daos_size_t file_size;
   DFSRandomAccessFile(std::string dfs_path, dfs_t* file_system, daos_handle_t eqh,
-                      dfs_obj_t* obj, size_t num_of_buffers)
+                      dfs_obj_t* obj)
       : dfs_path(std::move(dfs_path)) {
     daos_fs = file_system;
     daos_file.file = obj;
     dfs_get_size(daos_fs, obj, &file_size);
+    size_t num_of_buffers;
+    size_t buff_size;
+
+    if(char* env_num_of_buffers = std::getenv("TF_IO_DAOS_NUM_OF_BUFFERS")) {
+      num_of_buffers = atoi(env_num_of_buffers);
+    }
+    else {
+      num_of_buffers = NUM_OF_BUFFERS;
+    }
+
+    if(char* env_buff_size = std::getenv("TF_IO_DAOS_BUFFER_SIZE")) {
+      buff_size = GetStorageSize(env_buff_size);
+    }
+    else {
+      buff_size = BUFF_SIZE;
+    }
+    std::cout << buff_size / 1024 / 1024 << std::endl;
+    std::cout << num_of_buffers << std::endl;
     for(size_t i = 0; i < num_of_buffers; i++) {
-      buffers.push_back(ReadBuffer(i, eqh, BUFF_SIZE));
+      buffers.push_back(ReadBuffer(i, eqh, buff_size));
     }
   }
 } DFSRandomAccessFile;
@@ -37,21 +55,21 @@ void Cleanup(TF_RandomAccessFile* file) {
 
 int64_t Read(const TF_RandomAccessFile* file, uint64_t offset, size_t n,
              char* ret, TF_Status* status) {
-  int rc = 0;
   auto dfs_file = static_cast<DFSRandomAccessFile*>(file->plugin_file);
   for(auto& read_buf: dfs_file->buffers) {
     if(read_buf.CacheHit(offset, n))
       return read_buf.CopyFromCache(ret, offset, n, dfs_file->file_size, status);
   }
 
-  dfs_file->buffers[0].ReadSync(dfs_file->daos_fs, dfs_file->daos_file.file, offset);
   size_t curr_offset = offset + BUFF_SIZE;
   for(size_t i = 1; i < dfs_file->buffers.size(); i++) {
     if(curr_offset > dfs_file->file_size) break;
     dfs_file->buffers[i].ReadAsync(dfs_file->daos_fs, dfs_file->daos_file.file, curr_offset);
     curr_offset += BUFF_SIZE;
   }
 
+  dfs_file->buffers[0].ReadSync(dfs_file->daos_fs, dfs_file->daos_file.file, offset);
+
   return dfs_file->buffers[0].CopyFromCache(ret, offset, n, dfs_file->file_size, status);
 }
 
@@ -157,7 +175,6 @@ void NewFile(const TF_Filesystem* filesystem, const char* path, mode_t mode,
   std::string pool, cont, file_path;
   rc = daos->Setup(path, pool, cont, file_path, status);
   if (rc) return;
-
   daos->dfsNewFile(file_path, mode, flags, obj, status);
 }
 
@@ -188,7 +205,7 @@ void NewRandomAccessFile(const TF_Filesystem* filesystem, const char* path,
     return;
   }
   auto random_access_file =  new tf_random_access_file::DFSRandomAccessFile(path, daos->daos_fs, daos->mEventQueueHandle, 
-                                                                            obj, NUM_OF_BUFFERS);
+                                                                            obj);
   random_access_file->buffers[0].ReadAsync(daos->daos_fs, random_access_file->daos_file.file, 0);
   file->plugin_file = random_access_file;
   TF_SetStatus(status, TF_OK, "");

tensorflow_io/core/filesystems/dfs/dfs_utils.cc

@@ -1,6 +1,6 @@
 #include "tensorflow_io/core/filesystems/dfs/dfs_utils.h"
 
-std::string FormatStorageSize(uint64_t size) {
+std::string GetStorageString(uint64_t size) {
   if (size < KILO) {
     return std::to_string(size);
   } else if (size < MEGA) {
@@ -14,18 +14,55 @@ std::string FormatStorageSize(uint64_t size) {
   }
 }
 
+size_t GetStorageSize(std::string size) {
+  char size_char = size.back();
+  size_t curr_scale = 1;
+  switch (size_char)
+  {
+  case 'K':
+    size.pop_back();
+    curr_scale *= 1024;
+    return (size_t) atoi(size.c_str()) * curr_scale;
+    break;
+  case 'M':
+    size.pop_back();
+    curr_scale *= 1024 * 1024;
+    return (size_t) atoi(size.c_str()) * curr_scale;
+    break;
+  case 'G':
+    size.pop_back();
+    curr_scale *= 1024 * 1024 * 1024;
+    return (size_t) atoi(size.c_str()) * curr_scale;
+    break;
+  case 'T':
+    size.pop_back();
+    curr_scale *= 1024 * 1024 * 1024;
+    return (size_t) atoi(size.c_str()) * curr_scale * 1024;
+    break;
+  default:
+    return atoi(size.c_str());
+    break;
+  }
+}
+
 int ParseDFSPath(const std::string& path, std::string& pool_string,
                  std::string& cont_string, std::string& filename) {
   size_t pool_start = path.find("://") + 3;
   struct duns_attr_t* attr =
       (struct duns_attr_t*)malloc(sizeof(struct duns_attr_t));
   attr->da_rel_path = NULL;
-  attr->da_flags = 0;
-  attr->da_no_prefix = false;
-  std::string direct_path = "daos://" + path.substr(pool_start);
+  attr->da_flags = 1;
+  attr->da_no_prefix = true;
+  std::string direct_path = "/" + path.substr(pool_start);
   int rc = duns_resolve_path(direct_path.c_str(), attr);
-  if (rc) {
-    return rc;
+  if (rc == 2) {
+    attr->da_rel_path = NULL;
+    attr->da_flags = 0;
+    attr->da_no_prefix = false;
+    direct_path = "daos://" + path.substr(pool_start);
+    rc = duns_resolve_path(direct_path.c_str(), attr);
+    if(rc)
+      return rc;
   }
   pool_string = attr->da_pool;
   cont_string = attr->da_cont;
@@ -113,7 +150,7 @@ void DFS::Teardown() {
   do {
     ret = daos_eq_poll(mEventQueueHandle, 1, -1, 1, &(temp_event));
   } while(ret == 1); 
-  int rc = daos_eq_destroy(mEventQueueHandle, 0);
+  daos_eq_destroy(mEventQueueHandle, 0);
   Unmount();
   ClearConnections();
 }
@@ -198,15 +235,15 @@ int DFS::Query() {
               << std::endl;
     std::cout << "- SCM:" << std::endl;
     std::cout << "  Total size: "
-              << FormatStorageSize(pool_info.pi_space.ps_space.s_total[0]);
+              << GetStorageString(pool_info.pi_space.ps_space.s_total[0]);
     std::cout << "  Free: "
-              << FormatStorageSize(pool_info.pi_space.ps_space.s_free[0])
+              << GetStorageString(pool_info.pi_space.ps_space.s_free[0])
               << std::endl;
     std::cout << "- NVMe:" << std::endl;
     std::cout << "  Total size: "
-              << FormatStorageSize(pool_info.pi_space.ps_space.s_total[1]);
+              << GetStorageString(pool_info.pi_space.ps_space.s_total[1]);
     std::cout << "  Free: "
-              << FormatStorageSize(pool_info.pi_space.ps_space.s_free[1])
+              << GetStorageString(pool_info.pi_space.ps_space.s_free[1])
               << std::endl;
     std::cout << std::endl
               << "Connected Container: " << container.first << std::endl;
@@ -429,18 +466,15 @@ ReadBuffer::ReadBuffer(size_t id, daos_handle_t eqh, size_t size): id(id), buffe
     buffer = new char[size];
     buffer_offset = 0;
     event = new daos_event_t;
-    int rc = daos_event_init(event, eqh,nullptr);
-    if(rc) std::cout << "Failed to init" << std::endl;
+    daos_event_init(event, eqh,nullptr);
     valid = false;
 }
 
 ReadBuffer::~ReadBuffer() {
     if(event != nullptr) {
       bool event_status;
-      int rc = daos_event_test(event, 0, &event_status);
-      if(rc) std::cout << "Failed to Test while destroying" << std::endl;
-      rc = daos_event_fini(event);
-      if(rc) std::cout << "Failed to Finalize" << std::endl;
+      daos_event_test(event, 0, &event_status);
+      daos_event_fini(event);
     }
     delete [] buffer;
     delete event;
@@ -467,8 +501,7 @@ int
 ReadBuffer::WaitEvent() {
   if(valid) return 0; 
   bool event_status;
-  int rc = daos_event_test(event, -1, &event_status);
-  if(rc) std::cout << "Failed to Wait" << std::endl;
+  daos_event_test(event, -1, &event_status);
   if(event_status) {
     valid = true;
     return 0;
@@ -479,8 +512,7 @@ ReadBuffer::WaitEvent() {
 int
 ReadBuffer::AbortEvent() {
   bool event_status = false;
-  int rc = daos_event_test(event, 0, &event_status);
-  if(rc) std::cout << "Failed to check event status" << std::endl;
+  daos_event_test(event, 0, &event_status);
   if(!event_status)
     return daos_event_abort(event);
   else
@@ -491,23 +523,20 @@ int
 ReadBuffer::ReadAsync(dfs_t* daos_fs, dfs_obj_t* file, const size_t off) {
   int rc = AbortEvent();
   if(rc) return rc;
-  d_sg_list_t rsgl;
-  d_iov_t iov;
   d_iov_set(&iov, (void*)buffer, buffer_size);
   rsgl.sg_nr = 1;
   rsgl.sg_iovs = &iov;
   valid = false;
   buffer_offset = off;
   dfs_read(daos_fs, file, &rsgl,
            buffer_offset, &read_size, event);
+  return 0;
 }
 
 int
 ReadBuffer::ReadSync(dfs_t* daos_fs, dfs_obj_t* file, const size_t off) {
   int rc = AbortEvent();
   if(rc) return rc;
-  d_sg_list_t rsgl;
-  d_iov_t iov;
   d_iov_set(&iov, (void*)buffer, buffer_size);
   rsgl.sg_nr = 1;
   rsgl.sg_iovs = &iov;

tensorflow_io/core/filesystems/dfs/dfs_utils.h

@@ -9,8 +9,8 @@
 #define CONT_START 43
 #define PATH_START 80
 #define STACK 24
-#define NUM_OF_BUFFERS 2
-#define BUFF_SIZE 10*1024*1024
+#define NUM_OF_BUFFERS 256
+#define BUFF_SIZE 4*1024*1024
 
 #include <daos.h>
 #include <daos_fs.h>
@@ -130,7 +130,9 @@ typedef std::pair<std::string, daos_handle_t> id_handle_t;
 // enum used while wildcard matching
 enum Children_Status { NON_MATCHING, MATCHING_DIR, OK };
 
-std::string FormatStorageSize(uint64_t size);
+std::string GetStorageString(uint64_t size);
+
+size_t GetStorageSize(std::string size);
 
 // parse DFS path in the format of dfs://<pool_uuid>/<cont_uuid>/<filename>
 int ParseDFSPath(const std::string& path, std::string& pool_string,
@@ -240,6 +242,8 @@ class ReadBuffer {
     size_t buffer_size;
     daos_handle_t eqh;
     daos_event_t* event;
+    d_sg_list_t rsgl;
+    d_iov_t iov;
     bool  valid;
     daos_size_t read_size;
 };