Existing File Deletion when Opened in Write Mode

Omar Marzouk · Omar Marzouk · commit e3887d562c71 · 2022-05-06T11:46:06.000+02:00
diff --git a/tensorflow_io/core/filesystems/dfs/dfs_filesystem.cc b/tensorflow_io/core/filesystems/dfs/dfs_filesystem.cc
@@ -164,7 +164,7 @@ void Cleanup(TF_Filesystem* filesystem) {
   delete daos;
 }
 
-void NewFile(const TF_Filesystem* filesystem, const char* path, mode_t mode,
+void NewFile(const TF_Filesystem* filesystem, const char* path, File_Mode mode,
              int flags, dfs_obj_t** obj, TF_Status* status) {
   int rc;
   auto daos = static_cast<DFS*>(filesystem->plugin_filesystem)->Load();
@@ -181,8 +181,7 @@ void NewFile(const TF_Filesystem* filesystem, const char* path, mode_t mode,
 void NewWritableFile(const TF_Filesystem* filesystem, const char* path,
                      TF_WritableFile* file, TF_Status* status) {
   dfs_obj_t* obj = NULL;
-  NewFile(filesystem, path, S_IWUSR | S_IFREG, O_WRONLY | O_CREAT, &obj,
-          status);
+  NewFile(filesystem, path, WRITE, S_IWUSR | S_IFREG, &obj, status);
   if (TF_GetCode(status) != TF_OK) return;
   auto daos = static_cast<DFS*>(filesystem->plugin_filesystem)->Load();
   if (!daos) {
@@ -197,7 +196,7 @@ void NewWritableFile(const TF_Filesystem* filesystem, const char* path,
 void NewRandomAccessFile(const TF_Filesystem* filesystem, const char* path,
                          TF_RandomAccessFile* file, TF_Status* status) {
   dfs_obj_t* obj = NULL;
-  NewFile(filesystem, path, S_IRUSR | S_IFREG, O_RDONLY, &obj, status);
+  NewFile(filesystem, path, READ, S_IRUSR | S_IFREG, &obj, status);
   if (TF_GetCode(status) != TF_OK) return;
   auto daos = static_cast<DFS*>(filesystem->plugin_filesystem)->Load();
   if (!daos) {
@@ -215,8 +214,7 @@ void NewRandomAccessFile(const TF_Filesystem* filesystem, const char* path,
 void NewAppendableFile(const TF_Filesystem* filesystem, const char* path,
                        TF_WritableFile* file, TF_Status* status) {
   dfs_obj_t* obj = NULL;
-  NewFile(filesystem, path, S_IWUSR | S_IFREG, O_WRONLY | O_CREAT | O_APPEND,
-          &obj, status);
+  NewFile(filesystem, path, APPEND, S_IWUSR | S_IFREG, &obj, status);
   if (TF_GetCode(status) != TF_OK) return;
   auto daos = static_cast<DFS*>(filesystem->plugin_filesystem)->Load();
   if (!daos) {
@@ -296,41 +294,19 @@ void DeleteFileSystemEntry(const TF_Filesystem* filesystem, const char* path,
   int rc;
   std::string pool, cont, dir_path;
   auto daos = static_cast<DFS*>(filesystem->plugin_filesystem)->Load();
+
   if (!daos) {
     TF_SetStatus(status, TF_INTERNAL, "Error initializng DAOS API");
     return;
   }
-  rc = daos->Setup(path, pool, cont, dir_path, status);
-  if (rc) return;
 
-  dfs_obj_t* temp_obj;
-  rc = daos->dfsPathExists(dir_path, &temp_obj, 0);
-  if (rc) {
-    TF_SetStatus(status, TF_NOT_FOUND, "");
-    return;
-  }
-  if (!is_dir && S_ISDIR(temp_obj->mode)) {
-    TF_SetStatus(status, TF_FAILED_PRECONDITION, "");
-    return;
-  }
-  dfs_release(temp_obj);
-
-  size_t dir_start = dir_path.rfind("/") + 1;
-  std::string dir = dir_path.substr(dir_start);
-  dfs_obj_t* parent;
-  rc = daos->dfsFindParent(dir_path, &parent);
+  rc = daos->Setup(path, pool, cont, dir_path, status);
   if (rc) {
-    TF_SetStatus(status, TF_NOT_FOUND, "");
+    TF_SetStatus(status, TF_INTERNAL, "Error initializng DAOS API");
     return;
   }
 
-  rc = dfs_remove(daos->daos_fs, parent, dir.c_str(), recursive, NULL);
-  if (rc) {
-    TF_SetStatus(status, TF_INTERNAL, "Error Deleting Directory");
-  } else {
-    TF_SetStatus(status, TF_OK, "");
-  }
-  dfs_release(parent);
+  daos->dfsDeleteObject(dir_path, is_dir, recursive, status);
 }
 
 void DeleteSingleDir(const TF_Filesystem* filesystem, const char* path,
@@ -596,18 +572,21 @@ int GetChildren(const TF_Filesystem* filesystem, const char* path,
   rc = daos->dfsPathExists(dir_path, &obj, 0);
   if (rc) {
     TF_SetStatus(status, TF_NOT_FOUND, "");
+    dfs_release(obj);
     return -1;
   }
 
   if (!S_ISDIR(obj->mode)) {
     TF_SetStatus(status, TF_FAILED_PRECONDITION, "");
+    dfs_release(obj);
     return -1;
   }
 
   std::vector<std::string> children;
   rc = daos->dfsReadDir(obj, children);
   if (rc) {
     TF_SetStatus(status, TF_INTERNAL, "");
+    dfs_release(obj);
     return -1;
   }
 
diff --git a/tensorflow_io/core/filesystems/dfs/dfs_utils.cc b/tensorflow_io/core/filesystems/dfs/dfs_utils.cc
@@ -22,25 +22,33 @@ size_t GetStorageSize(std::string size) {
       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;
+  }
+}
+
+mode_t GetFlags(File_Mode mode) {
+  switch (mode) {
+    case READ:
+      return O_RDONLY;
+    case WRITE:
+      return O_WRONLY | O_CREAT;
+    case APPEND:
+      return O_WRONLY | O_APPEND | O_CREAT;
+    default:
+      return -1;
   }
 }
 
@@ -74,8 +82,8 @@ int ParseUUID(const std::string& str, uuid_t uuid) {
 }
 
 void CopyEntries(char*** entries, std::vector<std::string>& results) {
-  *entries = static_cast<char**>(tensorflow::io::plugin_memory_allocate(
-      results.size() * sizeof((*entries)[0])));
+  *entries = static_cast<char**>(
+      tensorflow::io::plugin_memory_allocate(results.size() * sizeof(char*)));
 
   for (uint32_t i = 0; i < results.size(); i++) {
     (*entries)[i] = static_cast<char*>(tensorflow::io::plugin_memory_allocate(
@@ -267,10 +275,47 @@ int DFS::ClearConnections() {
   return rc;
 }
 
-void DFS::dfsNewFile(std::string& file_path, mode_t mode, int flags,
+int DFS::dfsDeleteObject(std::string& dir_path, bool is_dir, bool recursive,
+                         TF_Status* status) {
+  dfs_obj_t* temp_obj;
+  int rc = dfsPathExists(dir_path, &temp_obj, 0);
+  if (rc) {
+    TF_SetStatus(status, TF_NOT_FOUND, "");
+    return -1;
+  }
+  if (!is_dir && S_ISDIR(temp_obj->mode)) {
+    TF_SetStatus(status, TF_FAILED_PRECONDITION, "");
+    return -1;
+  }
+  dfs_release(temp_obj);
+
+  size_t dir_start = dir_path.rfind("/") + 1;
+  std::string dir = dir_path.substr(dir_start);
+  dfs_obj_t* parent;
+  rc = dfsFindParent(dir_path, &parent);
+  if (rc) {
+    TF_SetStatus(status, TF_NOT_FOUND, "");
+    return -1;
+  }
+
+  rc = dfs_remove(daos_fs, parent, dir.c_str(), recursive, NULL);
+
+  dfs_release(parent);
+
+  if (rc) {
+    TF_SetStatus(status, TF_INTERNAL, "Error Deleting Existing Object");
+  } else {
+    TF_SetStatus(status, TF_OK, "");
+  }
+
+  return rc;
+}
+
+void DFS::dfsNewFile(std::string& file_path, File_Mode file_mode, int flags,
                      dfs_obj_t** obj, TF_Status* status) {
   int rc;
   dfs_obj_t* temp_obj;
+  mode_t open_flags;
   rc = dfsPathExists(file_path, &temp_obj, 0);
   if (rc && flags == O_RDONLY) {
     TF_SetStatus(status, TF_NOT_FOUND, "");
@@ -287,6 +332,13 @@ void DFS::dfsNewFile(std::string& file_path, mode_t mode, int flags,
     dfs_release(temp_obj);
   }
 
+  if (!rc && file_mode == WRITE) {
+    rc = dfsDeleteObject(file_path, false, false, status);
+    if (rc) return;
+  }
+
+  open_flags = GetFlags(file_mode);
+
   dfs_obj_t* parent;
   rc = dfsFindParent(file_path, &parent);
   if (rc) {
@@ -302,8 +354,8 @@ void DFS::dfsNewFile(std::string& file_path, mode_t mode, int flags,
   size_t file_start = file_path.rfind("/") + 1;
   std::string file_name = file_path.substr(file_start);
 
-  rc = dfs_open(daos_fs, parent, file_name.c_str(), mode, flags, 0, 0, NULL,
-                obj);
+  rc = dfs_open(daos_fs, parent, file_name.c_str(), flags, open_flags, 0, 0,
+                NULL, obj);
   if (rc) {
     TF_SetStatus(status, TF_INTERNAL, "Error Creating Writable File");
     return;
@@ -372,14 +424,17 @@ int DFS::dfsReadDir(dfs_obj_t* obj, std::vector<std::string>& children) {
   daos_anchor_t anchor = {0};
   uint32_t nr = STACK;
   struct dirent* dirs = (struct dirent*)malloc(nr * sizeof(struct dirent));
+  while (!daos_anchor_is_eof(&anchor)) {
+    rc = dfs_readdir(daos_fs, obj, &anchor, &nr, dirs);
+    if (rc) {
+      return rc;
+    }
 
-  rc = dfs_readdir(daos_fs, obj, &anchor, &nr, dirs);
-  if (rc) {
-    return rc;
-  }
-
-  for (uint32_t i = 0; i < nr; i++) {
-    children.push_back(dirs[i].d_name);
+    for (uint32_t i = 0; i < nr; i++) {
+      // std::cout << "Before File " << dirs[i].d_name << std::endl;
+      children.emplace_back(dirs[i].d_name);
+      // std::cout << "After File " << dirs[i].d_name  << std::endl;
+    }
   }
 
   free(dirs);
@@ -498,11 +553,11 @@ int ReadBuffer::WaitEvent() {
   if (valid) return 0;
   bool event_status;
   daos_event_test(event, -1, &event_status);
-  if (event_status) {
+  if (event_status && event->ev_error == DER_SUCCESS) {
     valid = true;
     return 0;
   }
-  return -1;
+  return event->ev_error;
 }
 
 int ReadBuffer::AbortEvent() {
@@ -535,7 +590,7 @@ int ReadBuffer::ReadSync(dfs_t* daos_fs, dfs_obj_t* file, const size_t off) {
   valid = false;
   buffer_offset = off;
   rc = dfs_read(daos_fs, file, &rsgl, off, &read_size, NULL);
-  valid = true;
+  if (!rc) valid = true;
   return rc;
 }
 
diff --git a/tensorflow_io/core/filesystems/dfs/dfs_utils.h b/tensorflow_io/core/filesystems/dfs/dfs_utils.h
@@ -127,8 +127,7 @@ typedef struct pool_info {
 
 typedef std::pair<std::string, daos_handle_t> id_handle_t;
 
-// enum used while wildcard matching
-enum Children_Status { NON_MATCHING, MATCHING_DIR, OK };
+enum File_Mode { READ, WRITE, APPEND };
 
 std::string GetStorageString(uint64_t size);
 
@@ -176,7 +175,7 @@ class DFS {
 
   int ClearConnections();
 
-  void dfsNewFile(std::string& file_path, mode_t mode, int flags,
+  void dfsNewFile(std::string& file_path, File_Mode mode, int flags,
                   dfs_obj_t** obj, TF_Status* status);
 
   int dfsPathExists(std::string& file, dfs_obj_t** obj, int release_obj = 1);
@@ -185,6 +184,9 @@ class DFS {
 
   int dfsCreateDir(std::string& dir_path, TF_Status* status);
 
+  int dfsDeleteObject(std::string& dir_path, bool is_dir, bool recursive,
+                      TF_Status* status);
+
   bool isRoot(std::string& file_path);
 
   int dfsReadDir(dfs_obj_t* obj, std::vector<std::string>& children);
