Implement a file factory mechanism to handle network file systems.

- Env dispatches to a FileSystem interface
- FileSystemFactory is used to look up the correct FileSystem implementation
based on the prefix of the filename
- Provide a registration mechanism to register different factories
Change: 119233977

tensorflow/core/kernels/immutable_constant_op_test.cc

@@ -54,18 +54,17 @@ class TestReadOnlyMemoryRegion : public ReadOnlyMemoryRegion {
   uint64 length_;
 };
 
-// A mock environment class that creates ReadOnlyMemoryRegion from allocated
-// memory.
-class TestEnvironment : public EnvWrapper {
+// A mock file system and environment class that creates ReadOnlyMemoryRegion
+// from allocated memory.
+class TestFileSystem : public NullFileSystem {
  public:
-  explicit TestEnvironment(Env* env) : EnvWrapper(env) {}
-  ~TestEnvironment() override = default;
+  ~TestFileSystem() override = default;
   Status NewReadOnlyMemoryRegionFromFile(
       const string& fname, ReadOnlyMemoryRegion** result) override {
     float val = 0;
     // For the tests create in-memory regions with float values equal to the
     // first letter of the region name.
-    switch (fname.front()) {
+    switch (GetNameFromURI(fname).front()) {
       case '2':
         val = 2.0f;
         break;
@@ -84,20 +83,23 @@ class TestEnvironment : public EnvWrapper {
   }
 };
 
+REGISTER_FILE_SYSTEM("test", TestFileSystem);
+
 struct ImmutableConstantOpTest {};
 
 TEST(ImmutableConstantOpTest, Simple) {
   const TensorShape kTestTensorShape({4, 1});
   const TensorShape kTestTensorShapeT({1, 4});
   GraphDefBuilder b(GraphDefBuilder::kFailImmediately);
-  Node* node1 = ops::ImmutableConst(DT_FLOAT, kTestTensorShape, "2", b.opts());
-  Node* node2 = ops::ImmutableConst(DT_FLOAT, kTestTensorShapeT, "3", b.opts());
+  Node* node1 =
+      ops::ImmutableConst(DT_FLOAT, kTestTensorShape, "test://2", b.opts());
+  Node* node2 =
+      ops::ImmutableConst(DT_FLOAT, kTestTensorShapeT, "test://3", b.opts());
   Node* result = ops::MatMul(node1, node2, b.opts());
   GraphDef graph_def;
   TF_ASSERT_OK(b.ToGraphDef(&graph_def));
-  std::unique_ptr<Env> env_ptr(new TestEnvironment(Env::Default()));
   SessionOptions session_options;
-  session_options.env = env_ptr.get();
+  session_options.env = Env::Default();
   session_options.config.mutable_graph_options()
       ->mutable_optimizer_options()
       ->set_opt_level(OptimizerOptions_Level_L0);
@@ -120,14 +122,15 @@ TEST(ImmutableConstantOpTest, ExecutionError) {
   const TensorShape kBadTensorShape({40, 100});
   const TensorShape kTestTensorShapeT({1, 4});
   GraphDefBuilder b(GraphDefBuilder::kFailImmediately);
-  Node* node1 = ops::ImmutableConst(DT_FLOAT, kBadTensorShape, "2", b.opts());
-  Node* node2 = ops::ImmutableConst(DT_FLOAT, kTestTensorShapeT, "3", b.opts());
+  Node* node1 =
+      ops::ImmutableConst(DT_FLOAT, kBadTensorShape, "test://2", b.opts());
+  Node* node2 =
+      ops::ImmutableConst(DT_FLOAT, kTestTensorShapeT, "test://3", b.opts());
   Node* result = ops::MatMul(node1, node2, b.opts());
   GraphDef graph_def;
   TF_ASSERT_OK(b.ToGraphDef(&graph_def));
-  std::unique_ptr<Env> env_ptr(new TestEnvironment(Env::Default()));
   SessionOptions session_options;
-  session_options.env = env_ptr.get();
+  session_options.env = Env::Default();
   std::unique_ptr<Session> session(NewSession(session_options));
   ASSERT_TRUE(session != nullptr) << "Failed to create session";
   TF_ASSERT_OK(session->Create(graph_def)) << "Can't create test graph";

tensorflow/core/platform/env.cc

@@ -15,6 +15,7 @@ limitations under the License.
 
 #include "tensorflow/core/platform/env.h"
 #include "tensorflow/core/lib/core/errors.h"
+#include "tensorflow/core/lib/gtl/map_util.h"
 #include "tensorflow/core/lib/gtl/stl_util.h"
 #include "tensorflow/core/platform/protobuf.h"
 
@@ -22,9 +23,92 @@ namespace tensorflow {
 
 Env::~Env() {}
 
-RandomAccessFile::~RandomAccessFile() {}
+Status Env::GetFileSystemForFile(const string& fname, FileSystem** result) {
+  string scheme = GetSchemeFromURI(fname);
+  FileSystem* file_system = GlobalFileSystemRegistry()->Lookup(scheme);
+  if (!file_system) {
+    return errors::Unimplemented("File system scheme ", scheme,
+                                 " not implemented");
+  }
+  *result = file_system;
+  return Status::OK();
+}
+
+Status Env::NewRandomAccessFile(const string& fname,
+                                RandomAccessFile** result) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(fname, &fs));
+  return fs->NewRandomAccessFile(fname, result);
+}
+
+Status Env::NewWritableFile(const string& fname, WritableFile** result) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(fname, &fs));
+  return fs->NewWritableFile(fname, result);
+}
+
+Status Env::NewAppendableFile(const string& fname, WritableFile** result) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(fname, &fs));
+  return fs->NewAppendableFile(fname, result);
+}
+
+Status Env::NewReadOnlyMemoryRegionFromFile(const string& fname,
+                                            ReadOnlyMemoryRegion** result) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(fname, &fs));
+  return fs->NewReadOnlyMemoryRegionFromFile(fname, result);
+}
+
+bool Env::FileExists(const string& fname) {
+  FileSystem* fs;
+  if (!GetFileSystemForFile(fname, &fs).ok()) {
+    return false;
+  }
+  return fs->FileExists(fname);
+}
 
-WritableFile::~WritableFile() {}
+Status Env::GetChildren(const string& dir, std::vector<string>* result) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(dir, &fs));
+  return fs->GetChildren(dir, result);
+}
+
+Status Env::DeleteFile(const string& fname) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(fname, &fs));
+  return fs->DeleteFile(fname);
+}
+
+Status Env::CreateDir(const string& dirname) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(dirname, &fs));
+  return fs->CreateDir(dirname);
+}
+
+Status Env::DeleteDir(const string& dirname) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(dirname, &fs));
+  return fs->DeleteDir(dirname);
+}
+
+Status Env::GetFileSize(const string& fname, uint64* file_size) {
+  FileSystem* fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(fname, &fs));
+  return fs->GetFileSize(fname, file_size);
+}
+
+Status Env::RenameFile(const string& src, const string& target) {
+  FileSystem* src_fs;
+  FileSystem* target_fs;
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(src, &src_fs));
+  TF_RETURN_IF_ERROR(GetFileSystemForFile(target, &target_fs));
+  if (src_fs != target_fs) {
+    return errors::Unimplemented("Renaming ", src, " to ", target,
+                                 " not implemented");
+  }
+  return src_fs->RenameFile(src, target);
+}
 
 Thread::~Thread() {}
 

tensorflow/core/platform/env.h

@@ -18,19 +18,20 @@ limitations under the License.
 
 #include <stdint.h>
 #include <string>
+#include <unordered_map>
 #include <vector>
+#include "tensorflow/core/lib/core/errors.h"
 #include "tensorflow/core/lib/core/status.h"
 #include "tensorflow/core/lib/core/stringpiece.h"
+#include "tensorflow/core/platform/file_system.h"
 #include "tensorflow/core/platform/macros.h"
+#include "tensorflow/core/platform/mutex.h"
 #include "tensorflow/core/platform/protobuf.h"
 #include "tensorflow/core/platform/types.h"
 
 namespace tensorflow {
 
-class RandomAccessFile;
-class ReadOnlyMemoryRegion;
 class Thread;
-class WritableFile;
 struct ThreadOptions;
 
 /// \brief An interface used by the tensorflow implementation to
@@ -55,6 +56,11 @@ class Env {
   /// The result of Default() belongs to this library and must never be deleted.
   static Env* Default();
 
+  /// \brief Returns the FileSystem object to handle operations on the file
+  /// specified by 'fname'. The FileSystem object is used as the implementation
+  /// for the file system related (non-virtual) functions that follow.
+  virtual Status GetFileSystemForFile(const string& fname, FileSystem** result);
+
   /// \brief Creates a brand new random access read-only file with the
   /// specified name.
 
@@ -64,8 +70,7 @@ class Env {
   /// status.
   ///
   /// The returned file may be concurrently accessed by multiple threads.
-  virtual Status NewRandomAccessFile(const string& fname,
-                                     RandomAccessFile** result) = 0;
+  Status NewRandomAccessFile(const string& fname, RandomAccessFile** result);
 
   /// \brief Creates an object that writes to a new file with the specified
   /// name.
@@ -76,8 +81,7 @@ class Env {
   /// returns non-OK.
   ///
   /// The returned file will only be accessed by one thread at a time.
-  virtual Status NewWritableFile(const string& fname,
-                                 WritableFile** result) = 0;
+  Status NewWritableFile(const string& fname, WritableFile** result);
 
   /// \brief Creates an object that either appends to an existing file, or
   /// writes to a new file (if the file does not exist to begin with).
@@ -87,8 +91,7 @@ class Env {
   /// non-OK.
   ///
   /// The returned file will only be accessed by one thread at a time.
-  virtual Status NewAppendableFile(const string& fname,
-                                   WritableFile** result) = 0;
+  Status NewAppendableFile(const string& fname, WritableFile** result);
 
   /// \brief Creates a readonly region of memory with the file context.
   ///
@@ -97,34 +100,33 @@ class Env {
   /// the caller. On failure stores nullptr in *result and returns non-OK.
   ///
   /// The returned memory region can be accessed from many threads in parallel.
-  virtual Status NewReadOnlyMemoryRegionFromFile(
-      const string& fname, ReadOnlyMemoryRegion** result) = 0;
+  Status NewReadOnlyMemoryRegionFromFile(const string& fname,
+                                         ReadOnlyMemoryRegion** result);
 
   /// Returns true iff the named file exists.
-  virtual bool FileExists(const string& fname) = 0;
+  bool FileExists(const string& fname);
 
   /// \brief Stores in *result the names of the children of the specified
   /// directory. The names are relative to "dir".
   ///
   /// Original contents of *results are dropped.
-  virtual Status GetChildren(const string& dir,
-                             std::vector<string>* result) = 0;
+  Status GetChildren(const string& dir, std::vector<string>* result);
 
   /// Deletes the named file.
-  virtual Status DeleteFile(const string& fname) = 0;
+  Status DeleteFile(const string& fname);
 
   /// Creates the specified directory.
-  virtual Status CreateDir(const string& dirname) = 0;
+  Status CreateDir(const string& dirname);
 
   /// Deletes the specified directory.
-  virtual Status DeleteDir(const string& dirname) = 0;
+  Status DeleteDir(const string& dirname);
 
   /// Stores the size of `fname` in `*file_size`.
-  virtual Status GetFileSize(const string& fname, uint64* file_size) = 0;
+  Status GetFileSize(const string& fname, uint64* file_size);
 
   /// \brief Renames file src to target. If target already exists, it will be
   /// replaced.
-  virtual Status RenameFile(const string& src, const string& target) = 0;
+  Status RenameFile(const string& src, const string& target);
 
   // TODO(jeff,sanjay): Add back thread/thread-pool support if needed.
   // TODO(jeff,sanjay): if needed, tighten spec so relative to epoch, or
@@ -184,68 +186,6 @@ class Env {
   void operator=(const Env&);
 };
 
-/// A file abstraction for randomly reading the contents of a file.
-class RandomAccessFile {
- public:
-  RandomAccessFile() {}
-  virtual ~RandomAccessFile();
-
-  /// \brief Reads up to `n` bytes from the file starting at `offset`.
-  ///
-  /// `scratch[0..n-1]` may be written by this routine.  Sets `*result`
-  /// to the data that was read (including if fewer than `n` bytes were
-  /// successfully read).  May set `*result` to point at data in
-  /// `scratch[0..n-1]`, so `scratch[0..n-1]` must be live when
-  /// `*result` is used.
-  ///
-  /// On OK returned status: `n` bytes have been stored in `*result`.
-  /// On non-OK returned status: `[0..n]` bytes have been stored in `*result`.
-  ///
-  /// Returns `OUT_OF_RANGE` if fewer than n bytes were stored in `*result`
-  /// because of EOF.
-  ///
-  /// Safe for concurrent use by multiple threads.
-  virtual Status Read(uint64 offset, size_t n, StringPiece* result,
-                      char* scratch) const = 0;
-
- private:
-  /// No copying allowed
-  RandomAccessFile(const RandomAccessFile&);
-  void operator=(const RandomAccessFile&);
-};
-
-/// \brief A file abstraction for sequential writing.
-///
-/// The implementation must provide buffering since callers may append
-/// small fragments at a time to the file.
-class WritableFile {
- public:
-  WritableFile() {}
-  virtual ~WritableFile();
-
-  virtual Status Append(const StringPiece& data) = 0;
-  virtual Status Close() = 0;
-  virtual Status Flush() = 0;
-  virtual Status Sync() = 0;
-
- private:
-  /// No copying allowed
-  WritableFile(const WritableFile&);
-  void operator=(const WritableFile&);
-};
-
-/// \brief A readonly memmapped file abstraction.
-///
-/// The implementation must guarantee that all memory is accessable when the
-/// object exists, independently from the Env that created it.
-class ReadOnlyMemoryRegion {
- public:
-  ReadOnlyMemoryRegion() {}
-  virtual ~ReadOnlyMemoryRegion() = default;
-  virtual const void* data() = 0;
-  virtual uint64 length() = 0;
-};
-
 /// \brief An implementation of Env that forwards all calls to another Env.
 ///
 /// May be useful to clients who wish to override just part of the
@@ -259,33 +199,11 @@ class EnvWrapper : public Env {
   /// Returns the target to which this Env forwards all calls
   Env* target() const { return target_; }
 
-  // The following text is boilerplate that forwards all methods to target()
-  Status NewRandomAccessFile(const string& f, RandomAccessFile** r) override {
-    return target_->NewRandomAccessFile(f, r);
-  }
-  Status NewWritableFile(const string& f, WritableFile** r) override {
-    return target_->NewWritableFile(f, r);
-  }
-  Status NewAppendableFile(const string& f, WritableFile** r) override {
-    return target_->NewAppendableFile(f, r);
-  }
-  Status NewReadOnlyMemoryRegionFromFile(
-      const string& fname, ReadOnlyMemoryRegion** result) override {
-    return target_->NewReadOnlyMemoryRegionFromFile(fname, result);
-  }
-  bool FileExists(const string& f) override { return target_->FileExists(f); }
-  Status GetChildren(const string& dir, std::vector<string>* r) override {
-    return target_->GetChildren(dir, r);
-  }
-  Status DeleteFile(const string& f) override { return target_->DeleteFile(f); }
-  Status CreateDir(const string& d) override { return target_->CreateDir(d); }
-  Status DeleteDir(const string& d) override { return target_->DeleteDir(d); }
-  Status GetFileSize(const string& f, uint64* s) override {
-    return target_->GetFileSize(f, s);
-  }
-  Status RenameFile(const string& s, const string& t) override {
-    return target_->RenameFile(s, t);
+  Status GetFileSystemForFile(const string& fname,
+                              FileSystem** result) override {
+    return target_->GetFileSystemForFile(fname, result);
   }
+
   uint64 NowMicros() override { return target_->NowMicros(); }
   void SleepForMicroseconds(int micros) override {
     target_->SleepForMicroseconds(micros);

tensorflow/core/platform/env_test.cc

@@ -71,4 +71,49 @@ TEST(EnvTest, FileToReadonlyMemoryRegion) {
   }
 }
 
+TEST(EnvTest, LocalFileSystem) {
+  // Test filename with file:// syntax.
+  Env* env = Env::Default();
+  const string dir = testing::TmpDir();
+  for (const int length : {0, 1, 1212, 2553, 4928, 8196, 9000, (1 << 20) - 1,
+                           1 << 20, (1 << 20) + 1}) {
+    string filename = io::JoinPath(dir, strings::StrCat("file", length));
+
+    filename = strings::StrCat("file://", filename);
+
+    // Write a file with the given length
+    const string input = CreateTestFile(env, filename, length);
+
+    // Read the file back and check equality
+    string output;
+    TF_CHECK_OK(ReadFileToString(env, filename, &output));
+    CHECK_EQ(length, output.size());
+    CHECK_EQ(input, output);
+  }
+}
+
+class InterPlanetaryFileSystem : public NullFileSystem {
+ public:
+  Status GetChildren(const string& dir, std::vector<string>* result) override {
+    std::vector<string> Planets = {"Mercury", "Venus",  "Earth",  "Mars",
+                                   "Jupiter", "Saturn", "Uranus", "Neptune"};
+    result->insert(result->end(), Planets.begin(), Planets.end());
+    return Status::OK();
+  }
+};
+
+REGISTER_FILE_SYSTEM("ipfs", InterPlanetaryFileSystem);
+
+TEST(EnvTest, IPFS) {
+  Env* env = Env::Default();
+  std::vector<string> planets;
+  TF_CHECK_OK(env->GetChildren("ipfs://solarsystem", &planets));
+  int c = 0;
+  std::vector<string> Planets = {"Mercury", "Venus",  "Earth",  "Mars",
+                                 "Jupiter", "Saturn", "Uranus", "Neptune"};
+  for (auto p : Planets) {
+    EXPECT_EQ(p, planets[c++]);
+  }
+}
+
 }  // namespace tensorflow

tensorflow/core/platform/file_system.cc

+/* Copyright 2015 Google Inc. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#include "tensorflow/core/platform/file_system.h"
+#include "tensorflow/core/lib/core/errors.h"
+#include "tensorflow/core/lib/gtl/map_util.h"
+#include "tensorflow/core/lib/gtl/stl_util.h"
+#include "tensorflow/core/lib/strings/str_util.h"
+#include "tensorflow/core/platform/protobuf.h"
+
+namespace tensorflow {
+
+FileSystem::~FileSystem() {}
+
+string FileSystem::TranslateName(const string& name) const { return name; }
+
+RandomAccessFile::~RandomAccessFile() {}
+
+WritableFile::~WritableFile() {}
+
+FileSystemRegistry* GlobalFileSystemRegistry() {
+  static FileSystemRegistry* registry = new FileSystemRegistry;
+  return registry;
+}
+
+void FileSystemRegistry::Register(const string& scheme,
+                                  FileSystemRegistry::Factory factory) {
+  mutex_lock lock(mu_);
+  QCHECK(!gtl::FindOrNull(registry_, scheme)) << "File factory for " << scheme
+                                              << " already registered";
+  registry_[scheme] = factory();
+}
+
+FileSystem* FileSystemRegistry::Lookup(const string& scheme) {
+  mutex_lock lock(mu_);
+  auto fs_ptr = gtl::FindOrNull(registry_, scheme);
+  if (!fs_ptr) {
+    return nullptr;
+  }
+  return *fs_ptr;
+}
+
+string GetSchemeFromURI(const string& name) {
+  auto colon_loc = name.find(":");
+  if (colon_loc != string::npos) {
+    return name.substr(0, colon_loc);
+  }
+  return "";
+}
+
+string GetNameFromURI(const string& name) {
+  string scheme = GetSchemeFromURI(name);
+  if (scheme == "") {
+    return name;
+  }
+  // Skip the 'scheme:' portion.
+  StringPiece filename{name.data() + scheme.length() + 1,
+                       name.length() - scheme.length() - 1};
+  // If the URI confirmed to scheme://filename, skip the two '/'s and return
+  // filename. Otherwise return the original 'name', and leave it up to the
+  // implementations to handle the full URI.
+  if (filename[0] == '/' && filename[1] == '/') {
+    return filename.substr(2).ToString();
+  }
+  return name;
+}
+
+}  // namespace tensorflow

tensorflow/core/platform/file_system.h

+/* Copyright 2015 Google Inc. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#ifndef TENSORFLOW_CORE_PLATFORM_FILE_SYSTEM_H_
+#define TENSORFLOW_CORE_PLATFORM_FILE_SYSTEM_H_
+
+#include <stdint.h>
+#include <functional>
+#include <string>
+#include <unordered_map>
+#include <vector>
+#include "tensorflow/core/lib/core/errors.h"
+#include "tensorflow/core/lib/core/status.h"
+#include "tensorflow/core/lib/core/stringpiece.h"
+#include "tensorflow/core/platform/macros.h"
+#include "tensorflow/core/platform/mutex.h"
+#include "tensorflow/core/platform/protobuf.h"
+#include "tensorflow/core/platform/types.h"
+
+namespace tensorflow {
+
+class RandomAccessFile;
+class ReadOnlyMemoryRegion;
+class WritableFile;
+
+/// An generic interface for accessing a file system.
+class FileSystem {
+ public:
+  FileSystem() {}
+
+  virtual ~FileSystem();
+
+  /// The following functions are the implementations used by the corresponding
+  /// functions in the Env class.
+  virtual Status NewRandomAccessFile(const string& fname,
+                                     RandomAccessFile** result) = 0;
+
+  virtual Status NewWritableFile(const string& fname,
+                                 WritableFile** result) = 0;
+
+  virtual Status NewAppendableFile(const string& fname,
+                                   WritableFile** result) = 0;
+
+  virtual Status NewReadOnlyMemoryRegionFromFile(
+      const string& fname, ReadOnlyMemoryRegion** result) = 0;
+
+  virtual bool FileExists(const string& fname) = 0;
+
+  virtual Status GetChildren(const string& dir,
+                             std::vector<string>* result) = 0;
+
+  virtual Status DeleteFile(const string& fname) = 0;
+
+  virtual Status CreateDir(const string& dirname) = 0;
+
+  virtual Status DeleteDir(const string& dirname) = 0;
+
+  virtual Status GetFileSize(const string& fname, uint64* file_size) = 0;
+
+  virtual Status RenameFile(const string& src, const string& target) = 0;
+
+  // Translate an URI to a filename usable by the FileSystem implementation. The
+  // implementation in this class returns the name as-is.
+  virtual string TranslateName(const string& name) const;
+};
+
+// Degenerate file system that provides no implementations.
+class NullFileSystem : public FileSystem {
+ public:
+  NullFileSystem() {}
+
+  ~NullFileSystem() override = default;
+
+  Status NewRandomAccessFile(const string& fname,
+                             RandomAccessFile** result) override {
+    return errors::Unimplemented("NewRandomAccessFile unimplemented");
+  }
+
+  Status NewWritableFile(const string& fname, WritableFile** result) override {
+    return errors::Unimplemented("NewWritableFile unimplemented");
+  }
+
+  Status NewAppendableFile(const string& fname,
+                           WritableFile** result) override {
+    return errors::Unimplemented("NewAppendableFile unimplemented");
+  }
+
+  Status NewReadOnlyMemoryRegionFromFile(
+      const string& fname, ReadOnlyMemoryRegion** result) override {
+    return errors::Unimplemented(
+        "NewReadOnlyMemoryRegionFromFile unimplemented");
+  }
+
+  bool FileExists(const string& fname) override { return false; }
+
+  Status GetChildren(const string& dir, std::vector<string>* result) override {
+    return errors::Unimplemented("GetChildren unimplemented");
+  }
+
+  Status DeleteFile(const string& fname) override {
+    return errors::Unimplemented("DeleteFile unimplemented");
+  }
+
+  Status CreateDir(const string& dirname) override {
+    return errors::Unimplemented("CreateDir unimplemented");
+  }
+
+  Status DeleteDir(const string& dirname) override {
+    return errors::Unimplemented("DeleteDir unimplemented");
+  }
+
+  Status GetFileSize(const string& fname, uint64* file_size) override {
+    return errors::Unimplemented("GetFileSize unimplemented");
+  }
+
+  Status RenameFile(const string& src, const string& target) override {
+    return errors::Unimplemented("RenameFile unimplemented");
+  }
+};
+
+/// A file abstraction for randomly reading the contents of a file.
+class RandomAccessFile {
+ public:
+  RandomAccessFile() {}
+  virtual ~RandomAccessFile();
+
+  /// \brief Reads up to `n` bytes from the file starting at `offset`.
+  ///
+  /// `scratch[0..n-1]` may be written by this routine.  Sets `*result`
+  /// to the data that was read (including if fewer than `n` bytes were
+  /// successfully read).  May set `*result` to point at data in
+  /// `scratch[0..n-1]`, so `scratch[0..n-1]` must be live when
+  /// `*result` is used.
+  ///
+  /// On OK returned status: `n` bytes have been stored in `*result`.
+  /// On non-OK returned status: `[0..n]` bytes have been stored in `*result`.
+  ///
+  /// Returns `OUT_OF_RANGE` if fewer than n bytes were stored in `*result`
+  /// because of EOF.
+  ///
+  /// Safe for concurrent use by multiple threads.
+  virtual Status Read(uint64 offset, size_t n, StringPiece* result,
+                      char* scratch) const = 0;
+
+ private:
+  /// No copying allowed
+  RandomAccessFile(const RandomAccessFile&);
+  void operator=(const RandomAccessFile&);
+};
+
+/// \brief A file abstraction for sequential writing.
+///
+/// The implementation must provide buffering since callers may append
+/// small fragments at a time to the file.
+class WritableFile {
+ public:
+  WritableFile() {}
+  virtual ~WritableFile();
+
+  virtual Status Append(const StringPiece& data) = 0;
+  virtual Status Close() = 0;
+  virtual Status Flush() = 0;
+  virtual Status Sync() = 0;
+
+ private:
+  /// No copying allowed
+  WritableFile(const WritableFile&);
+  void operator=(const WritableFile&);
+};
+
+/// \brief A readonly memmapped file abstraction.
+///
+/// The implementation must guarantee that all memory is accessable when the
+/// object exists, independently from the Env that created it.
+class ReadOnlyMemoryRegion {
+ public:
+  ReadOnlyMemoryRegion() {}
+  virtual ~ReadOnlyMemoryRegion() = default;
+  virtual const void* data() = 0;
+  virtual uint64 length() = 0;
+};
+
+/// \brief A registry for file system implementations.
+///
+/// Filenames are specified as an URI, which is of the form
+/// [scheme://]<filename>.
+/// File system implementations are registered using the REGISTER_FILE_SYSTEM
+/// macro, providing the 'scheme' as the key.
+class FileSystemRegistry {
+ public:
+  typedef std::function<FileSystem*()> Factory;
+
+  void Register(const string& scheme, Factory factory);
+  FileSystem* Lookup(const string& scheme);
+
+ private:
+  mutable mutex mu_;
+  mutable std::unordered_map<string, FileSystem*> registry_ GUARDED_BY(mu_);
+};
+
+FileSystemRegistry* GlobalFileSystemRegistry();
+
+namespace register_file_system {
+
+template <typename Factory>
+struct Register {
+  Register(const string& scheme) {
+    ::tensorflow::GlobalFileSystemRegistry()->Register(
+        scheme, []() -> FileSystem* { return new Factory; });
+  }
+};
+
+}  // namespace register_file_system
+
+// Given URI of the form [scheme://]<filename>, return 'scheme'.
+string GetSchemeFromURI(const string& name);
+
+// Given URI of the form [scheme://]<filename>, return 'filename'.
+string GetNameFromURI(const string& name);
+
+}  // namespace tensorflow
+
+// Register a FileSystem implementation for a scheme. Files with names that have
+// "scheme://" prefixes are routed to use this implementation.
+#define REGISTER_FILE_SYSTEM(scheme, factory) \
+  REGISTER_FILE_SYSTEM_UNIQ_HELPER(__COUNTER__, scheme, factory)
+#define REGISTER_FILE_SYSTEM_UNIQ_HELPER(ctr, scheme, factory) \
+  REGISTER_FILE_SYSTEM_UNIQ(ctr, scheme, factory)
+#define REGISTER_FILE_SYSTEM_UNIQ(ctr, scheme, factory)        \
+  static ::tensorflow::register_file_system::Register<factory> \
+      register_ff##ctr TF_ATTRIBUTE_UNUSED =                   \
+          ::tensorflow::register_file_system::Register<factory>(scheme)
+
+#endif  // TENSORFLOW_CORE_PLATFORM_FILE_SYSTEM_H_

tensorflow/core/platform/posix/env.cc

@@ -31,245 +31,12 @@ limitations under the License.
 #include "tensorflow/core/platform/env.h"
 #include "tensorflow/core/platform/load_library.h"
 #include "tensorflow/core/platform/logging.h"
+#include "tensorflow/core/platform/posix/posix_file_system.h"
 
 namespace tensorflow {
 
 namespace {
 
-error::Code ErrnoToCode(int err_number) {
-  error::Code code;
-  switch (err_number) {
-    case 0:
-      code = error::OK;
-      break;
-    case EINVAL:        // Invalid argument
-    case ENAMETOOLONG:  // Filename too long
-    case E2BIG:         // Argument list too long
-    case EDESTADDRREQ:  // Destination address required
-    case EDOM:          // Mathematics argument out of domain of function
-    case EFAULT:        // Bad address
-    case EILSEQ:        // Illegal byte sequence
-    case ENOPROTOOPT:   // Protocol not available
-    case ENOSTR:        // Not a STREAM
-    case ENOTSOCK:      // Not a socket
-    case ENOTTY:        // Inappropriate I/O control operation
-    case EPROTOTYPE:    // Protocol wrong type for socket
-    case ESPIPE:        // Invalid seek
-      code = error::INVALID_ARGUMENT;
-      break;
-    case ETIMEDOUT:  // Connection timed out
-    case ETIME:      // Timer expired
-      code = error::DEADLINE_EXCEEDED;
-      break;
-    case ENODEV:  // No such device
-    case ENOENT:  // No such file or directory
-    case ENXIO:   // No such device or address
-    case ESRCH:   // No such process
-      code = error::NOT_FOUND;
-      break;
-    case EEXIST:         // File exists
-    case EADDRNOTAVAIL:  // Address not available
-    case EALREADY:       // Connection already in progress
-      code = error::ALREADY_EXISTS;
-      break;
-    case EPERM:   // Operation not permitted
-    case EACCES:  // Permission denied
-    case EROFS:   // Read only file system
-      code = error::PERMISSION_DENIED;
-      break;
-    case ENOTEMPTY:   // Directory not empty
-    case EISDIR:      // Is a directory
-    case ENOTDIR:     // Not a directory
-    case EADDRINUSE:  // Address already in use
-    case EBADF:       // Invalid file descriptor
-    case EBUSY:       // Device or resource busy
-    case ECHILD:      // No child processes
-    case EISCONN:     // Socket is connected
-    case ENOTBLK:     // Block device required
-    case ENOTCONN:    // The socket is not connected
-    case EPIPE:       // Broken pipe
-    case ESHUTDOWN:   // Cannot send after transport endpoint shutdown
-    case ETXTBSY:     // Text file busy
-      code = error::FAILED_PRECONDITION;
-      break;
-    case ENOSPC:   // No space left on device
-    case EDQUOT:   // Disk quota exceeded
-    case EMFILE:   // Too many open files
-    case EMLINK:   // Too many links
-    case ENFILE:   // Too many open files in system
-    case ENOBUFS:  // No buffer space available
-    case ENODATA:  // No message is available on the STREAM read queue
-    case ENOMEM:   // Not enough space
-    case ENOSR:    // No STREAM resources
-    case EUSERS:   // Too many users
-      code = error::RESOURCE_EXHAUSTED;
-      break;
-    case EFBIG:      // File too large
-    case EOVERFLOW:  // Value too large to be stored in data type
-    case ERANGE:     // Result too large
-      code = error::OUT_OF_RANGE;
-      break;
-    case ENOSYS:           // Function not implemented
-    case ENOTSUP:          // Operation not supported
-    case EAFNOSUPPORT:     // Address family not supported
-    case EPFNOSUPPORT:     // Protocol family not supported
-    case EPROTONOSUPPORT:  // Protocol not supported
-    case ESOCKTNOSUPPORT:  // Socket type not supported
-    case EXDEV:            // Improper link
-      code = error::UNIMPLEMENTED;
-      break;
-    case EAGAIN:        // Resource temporarily unavailable
-    case ECONNREFUSED:  // Connection refused
-    case ECONNABORTED:  // Connection aborted
-    case ECONNRESET:    // Connection reset
-    case EINTR:         // Interrupted function call
-    case EHOSTDOWN:     // Host is down
-    case EHOSTUNREACH:  // Host is unreachable
-    case ENETDOWN:      // Network is down
-    case ENETRESET:     // Connection aborted by network
-    case ENETUNREACH:   // Network unreachable
-    case ENOLCK:        // No locks available
-    case ENOLINK:       // Link has been severed
-#if !defined(__APPLE__)
-    case ENONET:  // Machine is not on the network
-#endif
-      code = error::UNAVAILABLE;
-      break;
-    case EDEADLK:  // Resource deadlock avoided
-    case ESTALE:   // Stale file handle
-      code = error::ABORTED;
-      break;
-    case ECANCELED:  // Operation cancelled
-      code = error::CANCELLED;
-      break;
-    // NOTE: If you get any of the following (especially in a
-    // reproducible way) and can propose a better mapping,
-    // please email the owners about updating this mapping.
-    case EBADMSG:      // Bad message
-    case EIDRM:        // Identifier removed
-    case EINPROGRESS:  // Operation in progress
-    case EIO:          // I/O error
-    case ELOOP:        // Too many levels of symbolic links
-    case ENOEXEC:      // Exec format error
-    case ENOMSG:       // No message of the desired type
-    case EPROTO:       // Protocol error
-    case EREMOTE:      // Object is remote
-      code = error::UNKNOWN;
-      break;
-    default: {
-      code = error::UNKNOWN;
-      break;
-    }
-  }
-  return code;
-}
-
-static Status IOError(const string& context, int err_number) {
-  auto code = ErrnoToCode(err_number);
-  if (code == error::UNKNOWN) {
-    return Status(ErrnoToCode(err_number),
-                  context + "; " + strerror(err_number));
-  } else {
-    return Status(ErrnoToCode(err_number), context);
-  }
-}
-
-// pread() based random-access
-class PosixRandomAccessFile : public RandomAccessFile {
- private:
-  string filename_;
-  int fd_;
-
- public:
-  PosixRandomAccessFile(const string& fname, int fd)
-      : filename_(fname), fd_(fd) {}
-  ~PosixRandomAccessFile() override { close(fd_); }
-
-  Status Read(uint64 offset, size_t n, StringPiece* result,
-              char* scratch) const override {
-    Status s;
-    char* dst = scratch;
-    while (n > 0 && s.ok()) {
-      ssize_t r = pread(fd_, dst, n, static_cast<off_t>(offset));
-      if (r > 0) {
-        dst += r;
-        n -= r;
-        offset += r;
-      } else if (r == 0) {
-        s = Status(error::OUT_OF_RANGE, "Read less bytes than requested");
-      } else if (errno == EINTR || errno == EAGAIN) {
-        // Retry
-      } else {
-        s = IOError(filename_, errno);
-      }
-    }
-    *result = StringPiece(scratch, dst - scratch);
-    return s;
-  }
-};
-
-class PosixWritableFile : public WritableFile {
- private:
-  string filename_;
-  FILE* file_;
-
- public:
-  PosixWritableFile(const string& fname, FILE* f)
-      : filename_(fname), file_(f) {}
-
-  ~PosixWritableFile() override {
-    if (file_ != NULL) {
-      // Ignoring any potential errors
-      fclose(file_);
-    }
-  }
-
-  Status Append(const StringPiece& data) override {
-    size_t r = fwrite(data.data(), 1, data.size(), file_);
-    if (r != data.size()) {
-      return IOError(filename_, errno);
-    }
-    return Status::OK();
-  }
-
-  Status Close() override {
-    Status result;
-    if (fclose(file_) != 0) {
-      result = IOError(filename_, errno);
-    }
-    file_ = NULL;
-    return result;
-  }
-
-  Status Flush() override {
-    if (fflush(file_) != 0) {
-      return IOError(filename_, errno);
-    }
-    return Status::OK();
-  }
-
-  Status Sync() override {
-    Status s;
-    if (fflush(file_) != 0) {
-      s = IOError(filename_, errno);
-    }
-    return s;
-  }
-};
-
-class PosixReadOnlyMemoryRegion : public ReadOnlyMemoryRegion {
- public:
-  PosixReadOnlyMemoryRegion(const void* address, uint64 length)
-      : address_(address), length_(length) {}
-  ~PosixReadOnlyMemoryRegion() { munmap(const_cast<void*>(address_), length_); }
-  const void* data() override { return address_; }
-  uint64 length() override { return length_; }
-
- private:
-  const void* const address_;
-  const uint64 length_;
-};
-
 class StdThread : public Thread {
  public:
   // name and thread_options are both ignored.
@@ -288,131 +55,6 @@ class PosixEnv : public Env {
 
   ~PosixEnv() override { LOG(FATAL) << "Env::Default() must not be destroyed"; }
 
-  Status NewRandomAccessFile(const string& fname,
-                             RandomAccessFile** result) override {
-    *result = NULL;
-    Status s;
-    int fd = open(fname.c_str(), O_RDONLY);
-    if (fd < 0) {
-      s = IOError(fname, errno);
-    } else {
-      *result = new PosixRandomAccessFile(fname, fd);
-    }
-    return s;
-  }
-
-  Status NewWritableFile(const string& fname, WritableFile** result) override {
-    Status s;
-    FILE* f = fopen(fname.c_str(), "w");
-    if (f == NULL) {
-      *result = NULL;
-      s = IOError(fname, errno);
-    } else {
-      *result = new PosixWritableFile(fname, f);
-    }
-    return s;
-  }
-
-  Status NewAppendableFile(const string& fname,
-                           WritableFile** result) override {
-    Status s;
-    FILE* f = fopen(fname.c_str(), "a");
-    if (f == NULL) {
-      *result = NULL;
-      s = IOError(fname, errno);
-    } else {
-      *result = new PosixWritableFile(fname, f);
-    }
-    return s;
-  }
-
-  Status NewReadOnlyMemoryRegionFromFile(
-      const string& fname, ReadOnlyMemoryRegion** result) override {
-    *result = nullptr;
-    Status s = Status::OK();
-    int fd = open(fname.c_str(), O_RDONLY);
-    if (fd < 0) {
-      s = IOError(fname, errno);
-    } else {
-      struct stat st;
-      ::fstat(fd, &st);
-      const void* address =
-          mmap(nullptr, st.st_size, PROT_READ, MAP_PRIVATE, fd, 0);
-      if (address == MAP_FAILED) {
-        s = IOError(fname, errno);
-      } else {
-        *result = new PosixReadOnlyMemoryRegion(address, st.st_size);
-      }
-      close(fd);
-    }
-    return s;
-  }
-
-  bool FileExists(const string& fname) override {
-    return access(fname.c_str(), F_OK) == 0;
-  }
-
-  Status GetChildren(const string& dir, std::vector<string>* result) override {
-    result->clear();
-    DIR* d = opendir(dir.c_str());
-    if (d == NULL) {
-      return IOError(dir, errno);
-    }
-    struct dirent* entry;
-    while ((entry = readdir(d)) != NULL) {
-      StringPiece basename = entry->d_name;
-      if ((basename != ".") && (basename != "..")) {
-        result->push_back(entry->d_name);
-      }
-    }
-    closedir(d);
-    return Status::OK();
-  }
-
-  Status DeleteFile(const string& fname) override {
-    Status result;
-    if (unlink(fname.c_str()) != 0) {
-      result = IOError(fname, errno);
-    }
-    return result;
-  }
-
-  Status CreateDir(const string& name) override {
-    Status result;
-    if (mkdir(name.c_str(), 0755) != 0) {
-      result = IOError(name, errno);
-    }
-    return result;
-  }
-
-  Status DeleteDir(const string& name) override {
-    Status result;
-    if (rmdir(name.c_str()) != 0) {
-      result = IOError(name, errno);
-    }
-    return result;
-  }
-
-  Status GetFileSize(const string& fname, uint64* size) override {
-    Status s;
-    struct stat sbuf;
-    if (stat(fname.c_str(), &sbuf) != 0) {
-      *size = 0;
-      s = IOError(fname, errno);
-    } else {
-      *size = sbuf.st_size;
-    }
-    return s;
-  }
-
-  Status RenameFile(const string& src, const string& target) override {
-    Status result;
-    if (rename(src.c_str(), target.c_str()) != 0) {
-      result = IOError(src, errno);
-    }
-    return result;
-  }
-
   uint64 NowMicros() override {
     struct timeval tv;
     gettimeofday(&tv, NULL);
@@ -458,6 +100,8 @@ class PosixEnv : public Env {
 }  // namespace
 
 #if defined(PLATFORM_POSIX) || defined(__ANDROID__)
+REGISTER_FILE_SYSTEM("", PosixFileSystem);
+REGISTER_FILE_SYSTEM("file", LocalPosixFileSystem);
 Env* Env::Default() {
   static Env* default_env = new PosixEnv;
   return default_env;

tensorflow/core/platform/posix/posix_file_system.cc

+/* Copyright 2015 Google Inc. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#include <dirent.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <sys/mman.h>
+#include <sys/stat.h>
+#include <sys/time.h>
+#include <sys/types.h>
+#include <time.h>
+#include <unistd.h>
+
+#include "tensorflow/core/lib/core/error_codes.pb.h"
+#include "tensorflow/core/lib/strings/strcat.h"
+#include "tensorflow/core/platform/env.h"
+#include "tensorflow/core/platform/logging.h"
+#include "tensorflow/core/platform/posix/posix_file_system.h"
+
+namespace tensorflow {
+
+namespace {
+
+error::Code ErrnoToCode(int err_number) {
+  error::Code code;
+  switch (err_number) {
+    case 0:
+      code = error::OK;
+      break;
+    case EINVAL:        // Invalid argument
+    case ENAMETOOLONG:  // Filename too long
+    case E2BIG:         // Argument list too long
+    case EDESTADDRREQ:  // Destination address required
+    case EDOM:          // Mathematics argument out of domain of function
+    case EFAULT:        // Bad address
+    case EILSEQ:        // Illegal byte sequence
+    case ENOPROTOOPT:   // Protocol not available
+    case ENOSTR:        // Not a STREAM
+    case ENOTSOCK:      // Not a socket
+    case ENOTTY:        // Inappropriate I/O control operation
+    case EPROTOTYPE:    // Protocol wrong type for socket
+    case ESPIPE:        // Invalid seek
+      code = error::INVALID_ARGUMENT;
+      break;
+    case ETIMEDOUT:  // Connection timed out
+    case ETIME:      // Timer expired
+      code = error::DEADLINE_EXCEEDED;
+      break;
+    case ENODEV:  // No such device
+    case ENOENT:  // No such file or directory
+    case ENXIO:   // No such device or address
+    case ESRCH:   // No such process
+      code = error::NOT_FOUND;
+      break;
+    case EEXIST:         // File exists
+    case EADDRNOTAVAIL:  // Address not available
+    case EALREADY:       // Connection already in progress
+      code = error::ALREADY_EXISTS;
+      break;
+    case EPERM:   // Operation not permitted
+    case EACCES:  // Permission denied
+    case EROFS:   // Read only file system
+      code = error::PERMISSION_DENIED;
+      break;
+    case ENOTEMPTY:   // Directory not empty
+    case EISDIR:      // Is a directory
+    case ENOTDIR:     // Not a directory
+    case EADDRINUSE:  // Address already in use
+    case EBADF:       // Invalid file descriptor
+    case EBUSY:       // Device or resource busy
+    case ECHILD:      // No child processes
+    case EISCONN:     // Socket is connected
+    case ENOTBLK:     // Block device required
+    case ENOTCONN:    // The socket is not connected
+    case EPIPE:       // Broken pipe
+    case ESHUTDOWN:   // Cannot send after transport endpoint shutdown
+    case ETXTBSY:     // Text file busy
+      code = error::FAILED_PRECONDITION;
+      break;
+    case ENOSPC:   // No space left on device
+    case EDQUOT:   // Disk quota exceeded
+    case EMFILE:   // Too many open files
+    case EMLINK:   // Too many links
+    case ENFILE:   // Too many open files in system
+    case ENOBUFS:  // No buffer space available
+    case ENODATA:  // No message is available on the STREAM read queue
+    case ENOMEM:   // Not enough space
+    case ENOSR:    // No STREAM resources
+    case EUSERS:   // Too many users
+      code = error::RESOURCE_EXHAUSTED;
+      break;
+    case EFBIG:      // File too large
+    case EOVERFLOW:  // Value too large to be stored in data type
+    case ERANGE:     // Result too large
+      code = error::OUT_OF_RANGE;
+      break;
+    case ENOSYS:           // Function not implemented
+    case ENOTSUP:          // Operation not supported
+    case EAFNOSUPPORT:     // Address family not supported
+    case EPFNOSUPPORT:     // Protocol family not supported
+    case EPROTONOSUPPORT:  // Protocol not supported
+    case ESOCKTNOSUPPORT:  // Socket type not supported
+    case EXDEV:            // Improper link
+      code = error::UNIMPLEMENTED;
+      break;
+    case EAGAIN:        // Resource temporarily unavailable
+    case ECONNREFUSED:  // Connection refused
+    case ECONNABORTED:  // Connection aborted
+    case ECONNRESET:    // Connection reset
+    case EINTR:         // Interrupted function call
+    case EHOSTDOWN:     // Host is down
+    case EHOSTUNREACH:  // Host is unreachable
+    case ENETDOWN:      // Network is down
+    case ENETRESET:     // Connection aborted by network
+    case ENETUNREACH:   // Network unreachable
+    case ENOLCK:        // No locks available
+    case ENOLINK:       // Link has been severed
+#if !defined(__APPLE__)
+    case ENONET:  // Machine is not on the network
+#endif
+      code = error::UNAVAILABLE;
+      break;
+    case EDEADLK:  // Resource deadlock avoided
+    case ESTALE:   // Stale file handle
+      code = error::ABORTED;
+      break;
+    case ECANCELED:  // Operation cancelled
+      code = error::CANCELLED;
+      break;
+    // NOTE: If you get any of the following (especially in a
+    // reproducible way) and can propose a better mapping,
+    // please email the owners about updating this mapping.
+    case EBADMSG:      // Bad message
+    case EIDRM:        // Identifier removed
+    case EINPROGRESS:  // Operation in progress
+    case EIO:          // I/O error
+    case ELOOP:        // Too many levels of symbolic links
+    case ENOEXEC:      // Exec format error
+    case ENOMSG:       // No message of the desired type
+    case EPROTO:       // Protocol error
+    case EREMOTE:      // Object is remote
+      code = error::UNKNOWN;
+      break;
+    default: {
+      code = error::UNKNOWN;
+      break;
+    }
+  }
+  return code;
+}
+
+// pread() based random-access
+class PosixRandomAccessFile : public RandomAccessFile {
+ private:
+  string filename_;
+  int fd_;
+
+ public:
+  PosixRandomAccessFile(const string& fname, int fd)
+      : filename_(fname), fd_(fd) {}
+  ~PosixRandomAccessFile() override { close(fd_); }
+
+  Status Read(uint64 offset, size_t n, StringPiece* result,
+              char* scratch) const override {
+    Status s;
+    char* dst = scratch;
+    while (n > 0 && s.ok()) {
+      ssize_t r = pread(fd_, dst, n, static_cast<off_t>(offset));
+      if (r > 0) {
+        dst += r;
+        n -= r;
+        offset += r;
+      } else if (r == 0) {
+        s = Status(error::OUT_OF_RANGE, "Read less bytes than requested");
+      } else if (errno == EINTR || errno == EAGAIN) {
+        // Retry
+      } else {
+        s = IOError(filename_, errno);
+      }
+    }
+    *result = StringPiece(scratch, dst - scratch);
+    return s;
+  }
+};
+
+class PosixWritableFile : public WritableFile {
+ private:
+  string filename_;
+  FILE* file_;
+
+ public:
+  PosixWritableFile(const string& fname, FILE* f)
+      : filename_(fname), file_(f) {}
+
+  ~PosixWritableFile() override {
+    if (file_ != NULL) {
+      // Ignoring any potential errors
+      fclose(file_);
+    }
+  }
+
+  Status Append(const StringPiece& data) override {
+    size_t r = fwrite(data.data(), 1, data.size(), file_);
+    if (r != data.size()) {
+      return IOError(filename_, errno);
+    }
+    return Status::OK();
+  }
+
+  Status Close() override {
+    Status result;
+    if (fclose(file_) != 0) {
+      result = IOError(filename_, errno);
+    }
+    file_ = NULL;
+    return result;
+  }
+
+  Status Flush() override {
+    if (fflush(file_) != 0) {
+      return IOError(filename_, errno);
+    }
+    return Status::OK();
+  }
+
+  Status Sync() override {
+    Status s;
+    if (fflush(file_) != 0) {
+      s = IOError(filename_, errno);
+    }
+    return s;
+  }
+};
+
+class PosixReadOnlyMemoryRegion : public ReadOnlyMemoryRegion {
+ public:
+  PosixReadOnlyMemoryRegion(const void* address, uint64 length)
+      : address_(address), length_(length) {}
+  ~PosixReadOnlyMemoryRegion() { munmap(const_cast<void*>(address_), length_); }
+  const void* data() override { return address_; }
+  uint64 length() override { return length_; }
+
+ private:
+  const void* const address_;
+  const uint64 length_;
+};
+
+}  // namespace
+
+Status PosixFileSystem::NewRandomAccessFile(const string& fname,
+                                            RandomAccessFile** result) {
+  string translated_fname = TranslateName(fname);
+  *result = NULL;
+  Status s;
+  int fd = open(translated_fname.c_str(), O_RDONLY);
+  if (fd < 0) {
+    s = IOError(fname, errno);
+  } else {
+    *result = new PosixRandomAccessFile(translated_fname, fd);
+  }
+  return s;
+}
+
+Status PosixFileSystem::NewWritableFile(const string& fname,
+                                        WritableFile** result) {
+  string translated_fname = TranslateName(fname);
+  Status s;
+  FILE* f = fopen(translated_fname.c_str(), "w");
+  if (f == NULL) {
+    *result = NULL;
+    s = IOError(fname, errno);
+  } else {
+    *result = new PosixWritableFile(translated_fname, f);
+  }
+  return s;
+}
+
+Status PosixFileSystem::NewAppendableFile(const string& fname,
+                                          WritableFile** result) {
+  string translated_fname = TranslateName(fname);
+  Status s;
+  FILE* f = fopen(translated_fname.c_str(), "a");
+  if (f == NULL) {
+    *result = NULL;
+    s = IOError(fname, errno);
+  } else {
+    *result = new PosixWritableFile(translated_fname, f);
+  }
+  return s;
+}
+
+Status PosixFileSystem::NewReadOnlyMemoryRegionFromFile(
+    const string& fname, ReadOnlyMemoryRegion** result) {
+  string translated_fname = TranslateName(fname);
+  *result = nullptr;
+  Status s = Status::OK();
+  int fd = open(translated_fname.c_str(), O_RDONLY);
+  if (fd < 0) {
+    s = IOError(fname, errno);
+  } else {
+    struct stat st;
+    ::fstat(fd, &st);
+    const void* address =
+        mmap(nullptr, st.st_size, PROT_READ, MAP_PRIVATE, fd, 0);
+    if (address == MAP_FAILED) {
+      s = IOError(fname, errno);
+    } else {
+      *result = new PosixReadOnlyMemoryRegion(address, st.st_size);
+    }
+    close(fd);
+  }
+  return s;
+}
+
+bool PosixFileSystem::FileExists(const string& fname) {
+  return access(TranslateName(fname).c_str(), F_OK) == 0;
+}
+
+Status PosixFileSystem::GetChildren(const string& dir,
+                                    std::vector<string>* result) {
+  string translated_dir = TranslateName(dir);
+  result->clear();
+  DIR* d = opendir(translated_dir.c_str());
+  if (d == NULL) {
+    return IOError(dir, errno);
+  }
+  struct dirent* entry;
+  while ((entry = readdir(d)) != NULL) {
+    StringPiece basename = entry->d_name;
+    if ((basename != ".") && (basename != "..")) {
+      result->push_back(entry->d_name);
+    }
+  }
+  closedir(d);
+  return Status::OK();
+}
+
+Status PosixFileSystem::DeleteFile(const string& fname) {
+  Status result;
+  if (unlink(TranslateName(fname).c_str()) != 0) {
+    result = IOError(fname, errno);
+  }
+  return result;
+}
+
+Status PosixFileSystem::CreateDir(const string& name) {
+  Status result;
+  if (mkdir(TranslateName(name).c_str(), 0755) != 0) {
+    result = IOError(name, errno);
+  }
+  return result;
+}
+
+Status PosixFileSystem::DeleteDir(const string& name) {
+  Status result;
+  if (rmdir(TranslateName(name).c_str()) != 0) {
+    result = IOError(name, errno);
+  }
+  return result;
+}
+
+Status PosixFileSystem::GetFileSize(const string& fname, uint64* size) {
+  Status s;
+  struct stat sbuf;
+  if (stat(TranslateName(fname).c_str(), &sbuf) != 0) {
+    *size = 0;
+    s = IOError(fname, errno);
+  } else {
+    *size = sbuf.st_size;
+  }
+  return s;
+}
+
+Status PosixFileSystem::RenameFile(const string& src, const string& target) {
+  Status result;
+  if (rename(TranslateName(src).c_str(), TranslateName(target).c_str()) != 0) {
+    result = IOError(src, errno);
+  }
+  return result;
+}
+
+Status IOError(const string& context, int err_number) {
+  auto code = ErrnoToCode(err_number);
+  if (code == error::UNKNOWN) {
+    return Status(code, strings::StrCat(context, "; ", strerror(err_number)));
+  } else {
+    return Status(code, context);
+  }
+}
+
+}  // namespace tensorflow

tensorflow/core/platform/posix/posix_file_system.h

+/* Copyright 2015 Google Inc. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#ifndef TENSORFLOW_CORE_PLATFORM_POSIX_POSIX_FILE_SYSTEM_H_
+#define TENSORFLOW_CORE_PLATFORM_POSIX_POSIX_FILE_SYSTEM_H_
+
+#include "tensorflow/core/platform/env.h"
+
+namespace tensorflow {
+
+class PosixFileSystem : public FileSystem {
+ public:
+  PosixFileSystem() {}
+
+  ~PosixFileSystem() {}
+
+  Status NewRandomAccessFile(const string& fname,
+                             RandomAccessFile** result) override;
+
+  Status NewWritableFile(const string& fname, WritableFile** result) override;
+
+  Status NewAppendableFile(const string& fname, WritableFile** result) override;
+
+  Status NewReadOnlyMemoryRegionFromFile(
+      const string& fname, ReadOnlyMemoryRegion** result) override;
+
+  bool FileExists(const string& fname) override;
+
+  Status GetChildren(const string& dir, std::vector<string>* result) override;
+
+  Status DeleteFile(const string& fname) override;
+
+  Status CreateDir(const string& name) override;
+
+  Status DeleteDir(const string& name) override;
+
+  Status GetFileSize(const string& fname, uint64* size) override;
+
+  Status RenameFile(const string& src, const string& target) override;
+};
+
+Status IOError(const string& context, int err_number);
+
+class LocalPosixFileSystem : public PosixFileSystem {
+ public:
+  string TranslateName(const string& name) const override {
+    return GetNameFromURI(name);
+  }
+};
+
+}  // namespace tensorflow
+
+#endif  // TENSORFLOW_CORE_PLATFORM_POSIX_POSIX_FILE_SYSTEM_H_

tensorflow/python/BUILD

@@ -120,6 +120,36 @@ cc_library(
     ],
 )
 
+cc_binary(
+    name = "framework/test_file_system.so",
+    srcs = ["framework/test_file_system.cc"],
+    linkopts = select({
+        "//conditions:default": [
+            "-Wl,-Bsymbolic",
+            "-lm",
+        ],
+        "//tensorflow:darwin": [],
+    }),
+    linkshared = 1,
+    deps = [
+        "//google/protobuf",
+        "//tensorflow/core:framework_headers_lib",
+    ],
+)
+
+py_test(
+    name = "file_system_test",
+    size = "small",
+    srcs = ["framework/file_system_test.py"],
+    data = [":framework/test_file_system.so"],
+    main = "framework/file_system_test.py",
+    srcs_version = "PY2AND3",
+    deps = [
+        ":framework_test_lib",
+        "//tensorflow:tensorflow_py",
+    ],
+)
+
 py_test(
     name = "pywrap_status_test",
     size = "small",

tensorflow/python/framework/file_system_test.py

+# Copyright 2015 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =============================================================================
+"""Tests for functions."""
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import os
+
+import tensorflow as tf
+
+
+class FileSystemTest(tf.test.TestCase):
+
+  def setUp(self):
+    file_system_library = os.path.join(tf.resource_loader.get_data_files_path(),
+                                       "test_file_system.so")
+    tf.load_file_system_library(file_system_library)
+
+  def testBasic(self):
+    with self.test_session() as sess:
+      reader = tf.WholeFileReader("test_reader")
+      queue = tf.FIFOQueue(99, [tf.string], shapes=())
+      queue.enqueue_many([["test://foo"]]).run()
+      queue.close().run()
+      key, value = sess.run(reader.read(queue))
+    self.assertEqual(key, "test://foo")
+    self.assertEqual(value, "AAAAAAAAAA")
+
+
+if __name__ == "__main__":
+  tf.test.main()

tensorflow/python/framework/framework_lib.py

@@ -37,6 +37,7 @@
 @@get_default_graph
 @@reset_default_graph
 @@import_graph_def
+@@load_file_system_library
 @@load_op_library
 
 ## Graph collections

tensorflow/python/framework/load_library.py

@@ -92,3 +92,45 @@ def load_op_library(library_filename):
   with _OP_LIBRARY_MAP_LOCK:
     _OP_LIBRARY_MAP[library_filename] = module
   return module
+
+
+_FILE_SYSTEM_LIBRARY_MAP = {}
+_FILE_SYSTEM_LIBRARY_MAP_LOCK = threading.Lock()
+
+
+def load_file_system_library(library_filename):
+  """Loads a TensorFlow plugin, containing file system implementation.
+
+  Pass `library_filename` to a platform-specific mechanism for dynamically
+  loading a library. The rules for determining the exact location of the
+  library are platform-specific and are not documented here.
+
+  Args:
+    library_filename: Path to the plugin.
+      Relative or absolute filesystem path to a dynamic library file.
+
+  Returns:
+    None.
+
+  Raises:
+    RuntimeError: when unable to load the library.
+  """
+  status = py_tf.TF_NewStatus()
+  lib_handle = py_tf.TF_LoadLibrary(library_filename, status)
+  try:
+    error_code = py_tf.TF_GetCode(status)
+    if error_code != 0:
+      error_msg = compat.as_text(py_tf.TF_Message(status))
+      with _FILE_SYSTEM_LIBRARY_MAP_LOCK:
+        if (error_code == error_codes_pb2.ALREADY_EXISTS and
+            'has already been loaded' in error_msg and
+            library_filename in _FILE_SYSTEM_LIBRARY_MAP):
+          return
+      # pylint: disable=protected-access
+      raise errors._make_specific_exception(None, None, error_msg, error_code)
+      # pylint: enable=protected-access
+  finally:
+    py_tf.TF_DeleteStatus(status)
+
+  with _FILE_SYSTEM_LIBRARY_MAP_LOCK:
+    _FILE_SYSTEM_LIBRARY_MAP[library_filename] = lib_handle

tensorflow/python/framework/test_file_system.cc

+/* Copyright 2015 Google Inc. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#include "tensorflow/core/platform/file_system.h"
+
+namespace tensorflow {
+
+class TestRandomAccessFile : public RandomAccessFile {
+  // The filecontents is all A's
+  Status Read(uint64 offset, size_t n, StringPiece* result,
+              char* scratch) const override {
+    for (int i = 0; i < n; ++i) {
+      scratch[i] = 'A';
+    }
+    *result = StringPiece(scratch, n);
+    return Status::OK();
+  }
+};
+
+class TestFileSystem : public NullFileSystem {
+ public:
+  Status NewRandomAccessFile(const string& fname,
+                             RandomAccessFile** result) override {
+    *result = new TestRandomAccessFile;
+    return Status::OK();
+  }
+  // Always return size of 10
+  Status GetFileSize(const string& fname, uint64* file_size) override {
+    *file_size = 10;
+    return Status::OK();
+  }
+};
+
+REGISTER_FILE_SYSTEM("test", TestFileSystem);
+
+}  // namespace tensorflow