Asynchronous read ahed

Author: Omar Marzouk

tensorflow_io/core/filesystems/dfs/dfs_filesystem.cc

@@ -20,7 +20,7 @@ typedef struct DFSRandomAccessFile {
     daos_file.file = obj;
     dfs_get_size(daos_fs, obj, &file_size);
     for(size_t i = 0; i < num_of_buffers; i++) {
-      buffers.push_back(ReadBuffer(eqh, BUFF_SIZE));
+      buffers.push_back(ReadBuffer(i, eqh, BUFF_SIZE));
     }
   }
 } DFSRandomAccessFile;
@@ -39,47 +39,20 @@ 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);
-  int counter = 0;
   for(auto& read_buf: dfs_file->buffers) {
-    if(read_buf.CacheHit(offset, n)){
-      rc = read_buf.CopyData(ret, offset, n);
-      if (rc) {
-        TF_SetStatus(status, TF_INTERNAL, "");
-        return 0;
-      }
-
-      if (offset + n > dfs_file->file_size) {
-        TF_SetStatus(status, TF_OUT_OF_RANGE, "");
-        return dfs_file->file_size - offset;
-      }
-
-      TF_SetStatus(status, TF_OK, "");
-      return n;
-    }
-    counter++;
+    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);
-  rc = dfs_file->buffers[0].CopyData(ret, offset, n);
   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].ReadSync(dfs_file->daos_fs, dfs_file->daos_file.file, curr_offset);
+    dfs_file->buffers[i].ReadAsync(dfs_file->daos_fs, dfs_file->daos_file.file, curr_offset);
     curr_offset += BUFF_SIZE;
   }
 
-  if (rc) {
-    TF_SetStatus(status, TF_INTERNAL, "");
-    return 0;
-  }
-
-  if (offset + n > dfs_file->file_size) {
-    TF_SetStatus(status, TF_OUT_OF_RANGE, "");
-    return dfs_file->file_size - offset;
-  }
-
-  TF_SetStatus(status, TF_OK, "");
-  return n;
+  return dfs_file->buffers[0].CopyFromCache(ret, offset, n, dfs_file->file_size, status);
 }
 
 }  // namespace tf_random_access_file
@@ -216,7 +189,7 @@ void NewRandomAccessFile(const TF_Filesystem* filesystem, const char* path,
   }
   auto random_access_file =  new tf_random_access_file::DFSRandomAccessFile(path, daos->daos_fs, daos->mEventQueueHandle, 
                                                                             obj, NUM_OF_BUFFERS);
-  random_access_file->buffers[0].ReadSync(daos->daos_fs, random_access_file->daos_file.file, 0);
+  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, "");
 }
@@ -695,4 +668,4 @@ void ProvideFilesystemSupportFor(TF_FilesystemPluginOps* ops, const char* uri) {
 
 }  // namespace dfs
 }  // namespace io
-}  // namespace tensorflow
+}  // namespace tensorflow

tensorflow_io/core/filesystems/dfs/dfs_utils.cc

@@ -113,7 +113,7 @@ void DFS::Teardown() {
   do {
     ret = daos_eq_poll(mEventQueueHandle, 1, -1, 1, &(temp_event));
   } while(ret == 1); 
-  daos_eq_destroy(mEventQueueHandle, 0);
+  int rc = daos_eq_destroy(mEventQueueHandle, 0);
   Unmount();
   ClearConnections();
 }
@@ -425,19 +425,22 @@ int DFS::DisconnectContainer(std::string pool_string, std::string cont_string) {
   return rc;
 }
 
-ReadBuffer::ReadBuffer(daos_handle_t eqh, size_t size): buffer_size(size), eqh(eqh) {
+ReadBuffer::ReadBuffer(size_t id, daos_handle_t eqh, size_t size): id(id), buffer_size(size), eqh(eqh) {
     buffer = new char[size];
     buffer_offset = 0;
     event = new daos_event_t;
-    daos_event_init(event, eqh,nullptr);
+    int rc = daos_event_init(event, eqh,nullptr);
+    if(rc) std::cout << "Failed to init" << std::endl;
     valid = false;
 }
 
 ReadBuffer::~ReadBuffer() {
     if(event != nullptr) {
       bool event_status;
-      daos_event_test(event, 0, &event_status);
-      daos_event_fini(event);
+      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;
     }
     delete [] buffer;
     delete event;
@@ -449,22 +452,25 @@ ReadBuffer::ReadBuffer(ReadBuffer&& read_buffer) {
   buffer = std::move(read_buffer.buffer);
   event = std::move(read_buffer.event);
   buffer_offset = 0;
+  id = read_buffer.id;
   valid = false;
   read_buffer.buffer = nullptr;
   read_buffer.event = nullptr;
 }
 
 bool
-ReadBuffer::CacheHit(size_t pos, size_t len) {
-  return pos >= buffer_offset && len < buffer_size && (pos+len <= buffer_offset + buffer_size); 
+ReadBuffer::CacheHit(const size_t pos, const size_t len) {
+  return pos >= buffer_offset && len <= buffer_size && (pos+len <= buffer_offset + buffer_size); 
 }
 
 int
 ReadBuffer::WaitEvent() {
-  if(valid) return 0;
+  if(valid) return 0; 
   bool event_status;
-  daos_event_test(event, -1, &event_status);
+  int rc = daos_event_test(event, -1, &event_status);
+  if(rc) std::cout << "Failed to Wait" << std::endl;
   if(event_status) {
+    valid = true;
     return 0;
   }
   return -1;
@@ -473,39 +479,38 @@ ReadBuffer::WaitEvent() {
 int
 ReadBuffer::AbortEvent() {
   bool event_status = false;
-  daos_event_test(event, 0, &event_status);
+  int rc = daos_event_test(event, 0, &event_status);
+  if(rc) std::cout << "Failed to check event status" << std::endl;
   if(!event_status)
     return daos_event_abort(event);
   else
     return 0;
 }
 
 int
-ReadBuffer::ReadAsync(dfs_t* daos_fs, dfs_obj_t* file, size_t off) {
+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;
-  daos_size_t read_size;
   valid = false;
   buffer_offset = off;
   dfs_read(daos_fs, file, &rsgl,
            buffer_offset, &read_size, event);
 }
 
 int
-ReadBuffer::ReadSync(dfs_t* daos_fs, dfs_obj_t* file, size_t off) {
+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;
-  daos_size_t read_size;
   valid = false;
   buffer_offset = off;
   rc =  dfs_read(daos_fs, file, &rsgl,
@@ -515,27 +520,29 @@ ReadBuffer::ReadSync(dfs_t* daos_fs, dfs_obj_t* file, size_t off) {
 }
 
 int
-ReadBuffer::CopyData(char* ret, size_t off, size_t n) {
+ReadBuffer::CopyData(char* ret, const size_t off, const size_t n) {
   int rc = WaitEvent();
   if(rc) return rc;
-  memcpy(ret, buffer + (off - buffer_offset), n);
+  memcpy(ret, buffer + (off - buffer_offset), n); 
   return 0;
 }
 
 int
-ReadBuffer::CopyFromCache(char* ret, size_t off, size_t n, daos_size_t file_size, TF_Status* status){
-    int rc = CopyData(ret, off, n);
+ReadBuffer::CopyFromCache(char* ret, const size_t off, const size_t n, const daos_size_t file_size, TF_Status* status){
+    size_t read_size;
+    read_size = off + n > file_size? file_size - off : n;
+    int rc = CopyData(ret, off, read_size);
     if (rc) {
       TF_SetStatus(status, TF_INTERNAL, "");
       return 0;
     }
 
     if (off + n > file_size) {
       TF_SetStatus(status, TF_OUT_OF_RANGE, "");
-      return file_size - off;
+      return read_size;
     }
 
     TF_SetStatus(status, TF_OK, "");
-    return n;
+    return read_size;
 }
 

tensorflow_io/core/filesystems/dfs/dfs_utils.h

@@ -206,14 +206,14 @@ void CopyEntries(char*** entries, std::vector<std::string>& results);
 
 class ReadBuffer {
   public:
-  ReadBuffer(daos_handle_t eqh, size_t size);
+  ReadBuffer(size_t id, daos_handle_t eqh, size_t size);
 
   ReadBuffer(ReadBuffer&&);
 
   ~ReadBuffer();
 
   bool
-  CacheHit(size_t pos, size_t off);
+  CacheHit(const size_t pos, const size_t off);
 
   int
   WaitEvent();
@@ -222,18 +222,19 @@ class ReadBuffer {
   AbortEvent();
 
   int
-  ReadAsync(dfs_t* dfs, dfs_obj_t* file, size_t off);
+  ReadAsync(dfs_t* dfs, dfs_obj_t* file, const size_t off);
 
   int
-  ReadSync(dfs_t* dfs, dfs_obj_t* file, size_t off);
+  ReadSync(dfs_t* dfs, dfs_obj_t* file, const size_t off);
 
   int
-  CopyData(char* ret, size_t off, size_t n);
+  CopyData(char* ret, const size_t offset, const size_t n);
 
   int
-  CopyFromCache(char* ret, size_t off, size_t n, daos_size_t file_size, TF_Status* status);
+  CopyFromCache(char* ret, const size_t off, const size_t n, const daos_size_t file_size, TF_Status* status);
 
   private:
+    size_t id;
     char* buffer;
     size_t buffer_offset;
     size_t buffer_size;