Return finer-granularity status from Env::GetChildren*

Summary:
It'd be nice to use the error status type to distinguish
between user error and system error. For example, GetChildren can fail
listing a backup directory's contents either because a bad path was provided
(user error) or because an operation failed, e.g., a remote storage service
call failed (system error). In the former case, we want to continue and treat
the backup directory as empty; in the latter case, we want to immediately
propagate the error to the caller.

This diff uses NotFound to indicate user error and IOError to indicate
system error. Previously IOError indicated both.
Closes https://github.com/facebook/rocksdb/pull/1644

Differential Revision: D4312157

Pulled By: ajkr

fbshipit-source-id: 51b4f24

include/rocksdb/env.h

@@ -200,6 +200,10 @@ class Env {
   // Store in *result the names of the children of the specified directory.
   // The names are relative to "dir".
   // Original contents of *results are dropped.
+  // Returns OK if "dir" exists and "*result" contains its children.
+  //         NotFound if "dir" does not exist, the calling process does not have
+  //                  permission to access "dir", or if "dir" is invalid.
+  //         IOError if an IO Error was encountered
   virtual Status GetChildren(const std::string& dir,
                              std::vector<std::string>* result) = 0;
 
@@ -209,6 +213,10 @@ class Env {
   // result.
   // The name attributes are relative to "dir".
   // Original contents of *results are dropped.
+  // Returns OK if "dir" exists and "*result" contains its children.
+  //         NotFound if "dir" does not exist, the calling process does not have
+  //                  permission to access "dir", or if "dir" is invalid.
+  //         IOError if an IO Error was encountered
   virtual Status GetChildrenFileAttributes(const std::string& dir,
                                            std::vector<FileAttributes>* result);
 

port/win/env_win.cc

@@ -370,7 +370,14 @@ Status WinEnvIO::GetChildren(const std::string& dir,
     CloseDir);
 
   if (!dirp) {
-    status = IOError(dir, errno);
+    switch (errno) {
+      case EACCES:
+      case ENOENT:
+      case ENOTDIR:
+        return Status::NotFound();
+      default:
+        return IOError(dir, errno);
+    }
   } else {
     if (result->capacity() > 0) {
       output.reserve(result->capacity());

util/env_basic_test.cc

@@ -202,6 +202,8 @@ TEST_P(EnvBasicTestWithParam, Basics) {
   ASSERT_EQ(Status::NotFound(), env_->FileExists(test_dir_ + "/g"));
   ASSERT_OK(env_->GetChildren(test_dir_, &children));
   ASSERT_EQ(0U, children.size());
+  ASSERT_TRUE(
+      env_->GetChildren(test_dir_ + "/non_existent", &children).IsNotFound());
 }
 
 TEST_P(EnvBasicTestWithParam, ReadWrite) {

util/env_hdfs.cc

@@ -491,7 +491,7 @@ Status HdfsEnv::GetChildren(const std::string& path,
     break;
   }
   case HDFS_DOESNT_EXIST:  // directory does not exist, exit
-    break;
+    return Status::NotFound();
   default:          // anything else should be an error
     Log(InfoLogLevel::FATAL_LEVEL, mylog,
         "GetChildren hdfsExists call failed");

util/env_posix.cc

@@ -418,7 +418,14 @@ class PosixEnv : public Env {
     result->clear();
     DIR* d = opendir(dir.c_str());
     if (d == nullptr) {
-      return IOError(dir, errno);
+      switch (errno) {
+        case EACCES:
+        case ENOENT:
+        case ENOTDIR:
+          return Status::NotFound();
+        default:
+          return IOError(dir, errno);
+      }
     }
     struct dirent* entry;
     while ((entry = readdir(d)) != nullptr) {

util/memenv.cc

@@ -325,16 +325,21 @@ class InMemoryEnv : public EnvWrapper {
     MutexLock lock(&mutex_);
     result->clear();
 
+    bool found_dir = false;
     for (FileSystem::iterator i = file_map_.begin(); i != file_map_.end(); ++i){
       const std::string& filename = i->first;
 
-      if (filename.size() >= dir.size() + 1 && filename[dir.size()] == '/' &&
+      if (dir == filename) {
-          Slice(filename).starts_with(Slice(dir))) {
+        found_dir = true;
+      } else if (filename.size() >= dir.size() + 1 &&
+                 filename[dir.size()] == '/' &&
+                 Slice(filename).starts_with(Slice(dir))) {
+        found_dir = true;
         result->push_back(filename.substr(dir.size() + 1));
       }
     }
 
-    return Status::OK();
+    return found_dir ? Status::OK() : Status::NotFound();
   }
 
   void DeleteFileInternal(const std::string& fname) {
@@ -358,15 +363,24 @@ class InMemoryEnv : public EnvWrapper {
   }
 
   virtual Status CreateDir(const std::string& dirname) override {
+    auto ndirname = NormalizeFileName(dirname);
+    if (file_map_.find(ndirname) == file_map_.end()) {
+      FileState* file = new FileState();
+      file->Ref();
+      file_map_[ndirname] = file;
+    } else {
+      return Status::IOError();
+    }
     return Status::OK();
   }
 
   virtual Status CreateDirIfMissing(const std::string& dirname) override {
+    CreateDir(dirname);
     return Status::OK();
   }
 
   virtual Status DeleteDir(const std::string& dirname) override {
-    return Status::OK();
+    return DeleteFile(dirname);
   }
 
   virtual Status GetFileSize(const std::string& fname,

util/mock_env.cc

@@ -483,14 +483,18 @@ Status MockEnv::FileExists(const std::string& fname) {
 Status MockEnv::GetChildren(const std::string& dir,
                                std::vector<std::string>* result) {
   auto d = NormalizePath(dir);
+  bool found_dir = false;
   {
     MutexLock lock(&mutex_);
     result->clear();
     for (const auto& iter : file_map_) {
       const std::string& filename = iter.first;
 
-      if (filename.size() >= d.size() + 1 && filename[d.size()] == '/' &&
+      if (filename == d) {
-          Slice(filename).starts_with(Slice(d))) {
+        found_dir = true;
+      } else if (filename.size() >= d.size() + 1 && filename[d.size()] == '/' &&
+                 Slice(filename).starts_with(Slice(d))) {
+        found_dir = true;
         size_t next_slash = filename.find('/', d.size() + 1);
         if (next_slash != std::string::npos) {
           result->push_back(filename.substr(
@@ -502,7 +506,7 @@ Status MockEnv::GetChildren(const std::string& dir,
     }
   }
   result->erase(std::unique(result->begin(), result->end()), result->end());
-  return Status::OK();
+  return found_dir ? Status::OK() : Status::NotFound();
 }
 
 void MockEnv::DeleteFileInternal(const std::string& fname) {
@@ -526,15 +530,24 @@ Status MockEnv::DeleteFile(const std::string& fname) {
 }
 
 Status MockEnv::CreateDir(const std::string& dirname) {
+  auto dn = NormalizePath(dirname);
+  if (file_map_.find(dn) == file_map_.end()) {
+    MemFile* file = new MemFile(this, dn, false);
+    file->Ref();
+    file_map_[dn] = file;
+  } else {
+    return Status::IOError();
+  }
   return Status::OK();
 }
 
 Status MockEnv::CreateDirIfMissing(const std::string& dirname) {
+  CreateDir(dirname);
   return Status::OK();
 }
 
 Status MockEnv::DeleteDir(const std::string& dirname) {
-  return Status::OK();
+  return DeleteFile(dirname);
 }
 
 Status MockEnv::GetFileSize(const std::string& fname, uint64_t* file_size) {