Merge branch 'hotfix/0.9_closerace' into release/0.9

.travis.yml

@@ -19,7 +19,7 @@ addons:
     - libfuse-dev
 install:
 # Workaround homebrew bug, see https://twitter.com/axccl/status/1083393735277363205 and https://github.com/openPMD/openPMD-api/pull/431/files
-- if [ "${TRAVIS_OS_NAME}" == "osx" ]; then travis_wait brew upgrade --cleanup; travis_wait brew upgrade --cleanup; fi
+- if [ "${TRAVIS_OS_NAME}" == "osx" ]; then travis_wait brew upgrade; travis_wait brew upgrade; fi
 # Use new clang
 - if [ "${TRAVIS_OS_NAME}" == "linux" ] && [ "$CXX" = "clang++" ]; then export CXX="clang++-3.7" CC="clang-3.7"; fi
 # Detect number of CPU cores

ChangeLog.txt

@@ -1,3 +1,8 @@
+Version 0.9.11 (unreleased)
+--------------
+Fixed bugs:
+* Fix a race condition when a file descriptor is closed while there's read/write requests for that file being processed.
+
 Version 0.9.10
 --------------
 Fixed bugs:

src/fspp/impl/FilesystemImpl.cpp

@@ -131,7 +131,9 @@ int FilesystemImpl::openFile(File *file, int flags) {
 
 void FilesystemImpl::flush(int descriptor) {
   PROFILE(_flushNanosec);
-  _open_files.get(descriptor)->flush();
+  _open_files.load(descriptor, [](OpenFile* openFile) {
+	  openFile->flush();
+  });
 }
 
 void FilesystemImpl::closeFile(int descriptor) {
@@ -151,7 +153,9 @@ void FilesystemImpl::lstat(const bf::path &path, struct ::stat *stbuf) {
 
 void FilesystemImpl::fstat(int descriptor, struct ::stat *stbuf) {
   PROFILE(_fstatNanosec);
-  _open_files.get(descriptor)->stat(stbuf);
+  _open_files.load(descriptor, [&](OpenFile* openFile) {
+	  openFile->stat(stbuf);
+  });
 }
 
 void FilesystemImpl::chmod(const boost::filesystem::path &path, mode_t mode) {
@@ -181,27 +185,37 @@ void FilesystemImpl::truncate(const bf::path &path, off_t size) {
 
 void FilesystemImpl::ftruncate(int descriptor, off_t size) {
   PROFILE(_ftruncateNanosec);
-  _open_files.get(descriptor)->truncate(size);
+  _open_files.load(descriptor, [size] (OpenFile* openFile) {
+	  openFile->truncate(size);
+  });
 }
 
 size_t FilesystemImpl::read(int descriptor, void *buf, size_t count, off_t offset) {
   PROFILE(_readNanosec);
-  return _open_files.get(descriptor)->read(buf, count, offset);
+  return _open_files.load(descriptor, [buf, count, offset] (OpenFile* openFile) {
+	  return openFile->read(buf, count, offset);
+  });
 }
 
 void FilesystemImpl::write(int descriptor, const void *buf, size_t count, off_t offset) {
   PROFILE(_writeNanosec);
-  _open_files.get(descriptor)->write(buf, count, offset);
+  return _open_files.load(descriptor, [buf, count, offset] (OpenFile* openFile) {
+	  return openFile->write(buf, count, offset);
+  });
 }
 
 void FilesystemImpl::fsync(int descriptor) {
   PROFILE(_fsyncNanosec);
-  _open_files.get(descriptor)->fsync();
+  _open_files.load(descriptor, [] (OpenFile* openFile) {
+	  openFile->fsync();
+  });
 }
 
 void FilesystemImpl::fdatasync(int descriptor) {
   PROFILE(_fdatasyncNanosec);
-  _open_files.get(descriptor)->fdatasync();
+  _open_files.load(descriptor, [] (OpenFile* openFile) {
+	  openFile->fdatasync();
+  });
 }
 
 void FilesystemImpl::access(const bf::path &path, int mask) {

src/fspp/impl/FuseOpenFileList.h

@@ -4,10 +4,27 @@
 
 #include "../fs_interface/File.h"
 #include "../fs_interface/OpenFile.h"
+#include "../fuse/FuseErrnoException.h"
 #include <cpp-utils/macros.h>
+#include <cpp-utils/assert/assert.h>
 #include "IdList.h"
+#include <condition_variable>
 
 namespace fspp {
+namespace detail {
+class OnScopeExit final {
+public:
+	explicit OnScopeExit(std::function<void()> handler)
+		: _handler(std::move(handler)) {}
+
+	~OnScopeExit() {
+		_handler();
+	}
+
+private:
+	std::function<void()> _handler;
+};
+}
 
 class FuseOpenFileList final {
 public:
@@ -15,33 +32,84 @@ public:
   ~FuseOpenFileList();
 
   int open(cpputils::unique_ref<OpenFile> file);
-  OpenFile *get(int descriptor);
+  template<class Func>
+  auto load(int descriptor, Func&& callback);
   void close(int descriptor);
 
 private:
   IdList<OpenFile> _open_files;
 
+  std::unordered_map<int, size_t> _refcounts;
+  std::mutex _mutex;
+
+  std::condition_variable _refcount_zero_cv;
+
   DISALLOW_COPY_AND_ASSIGN(FuseOpenFileList);
 };
 
 inline FuseOpenFileList::FuseOpenFileList()
-  :_open_files() {
+  :_open_files(), _refcounts(), _mutex(), _refcount_zero_cv() {
 }
 
 inline FuseOpenFileList::~FuseOpenFileList() {
+	std::unique_lock<std::mutex> lock(_mutex);
+
+	// Wait until all pending requests are done
+	_refcount_zero_cv.wait(lock, [&] {
+		for (const auto& refcount : _refcounts) {
+			if (0 != refcount.second) {
+				return false;
+			}
+		}
+		return true;
+	});
+
+	// There might still be open files when the file system is shutdown, so we can't assert it's empty.
+	// But to check that _refcounts has been updated correctly, we can assert the invariant that we have as many
+	// refcounts as open files.
+	ASSERT(_refcounts.size() == _refcounts.size(), "Didn't clean up refcounts properly");
 }
 
 inline int FuseOpenFileList::open(cpputils::unique_ref<OpenFile> file) {
-  return _open_files.add(std::move(file));
+  std::lock_guard<std::mutex> lock(_mutex);
+
+  int descriptor = _open_files.add(std::move(file));
+  _refcounts.emplace(descriptor, 0);
+  return descriptor;
 }
 
-inline OpenFile *FuseOpenFileList::get(int descriptor) {
-  return _open_files.get(descriptor);
+template<class Func>
+inline auto FuseOpenFileList::load(int descriptor, Func&& callback) {
+  try {
+    std::unique_lock<std::mutex> lock(_mutex);
+	_refcounts.at(descriptor) += 1;
+	detail::OnScopeExit _([&] {
+		if (!lock.owns_lock()) { // own_lock can be true when _open_files.get() below fails before the lock is unlocked
+		  lock.lock();
+		}
+		_refcounts.at(descriptor) -= 1;
+		_refcount_zero_cv.notify_all();
+	});
+
+	OpenFile* loaded = _open_files.get(descriptor);
+	lock.unlock();
+
+	return std::forward<Func>(callback)(loaded);
+  } catch (const std::out_of_range& e) {
+    throw fspp::fuse::FuseErrnoException(EBADF);
+  }
 }
 
 inline void FuseOpenFileList::close(int descriptor) {
+  std::unique_lock<std::mutex> lock(_mutex);
+
+  _refcount_zero_cv.wait(lock, [&] {
+	  return 0 == _refcounts.at(descriptor);
+  });
+
   //The destructor of the stored FuseOpenFile closes the file
   _open_files.remove(descriptor);
+  _refcounts.erase(descriptor);
 }
 
 }

src/fspp/impl/IdList.h

@@ -2,7 +2,7 @@
 #ifndef MESSMER_FSPP_IMPL_IDLIST_H_
 #define MESSMER_FSPP_IMPL_IDLIST_H_
 
-#include <map>
+#include <unordered_map>
 #include <mutex>
 #include <stdexcept>
 #include <cpp-utils/pointer/unique_ref.h>
@@ -19,17 +19,18 @@ public:
   Entry *get(int id);
   const Entry *get(int id) const;
   void remove(int id);
+  size_t size() const;
+
 private:
-  std::map<int, cpputils::unique_ref<Entry>> _entries;
+  std::unordered_map<int, cpputils::unique_ref<Entry>> _entries;
   int _id_counter;
-  mutable std::mutex _mutex;
 
   DISALLOW_COPY_AND_ASSIGN(IdList<Entry>);
 };
 
 template<class Entry>
 IdList<Entry>::IdList()
-  : _entries(), _id_counter(0), _mutex() {
+  : _entries(), _id_counter(0) {
 }
 
 template<class Entry>
@@ -38,10 +39,9 @@ IdList<Entry>::~IdList() {
 
 template<class Entry>
 int IdList<Entry>::add(cpputils::unique_ref<Entry> entry) {
-  std::lock_guard<std::mutex> lock(_mutex);
   //TODO Reuse IDs (ids = descriptors)
   int new_id = ++_id_counter;
-  _entries.insert(std::make_pair(new_id, std::move(entry)));
+  _entries.emplace(new_id, std::move(entry));
   return new_id;
 }
 
@@ -52,14 +52,12 @@ Entry *IdList<Entry>::get(int id) {
 
 template<class Entry>
 const Entry *IdList<Entry>::get(int id) const {
-  std::lock_guard<std::mutex> lock(_mutex);
   const Entry *result = _entries.at(id).get();
   return result;
 }
 
 template<class Entry>
 void IdList<Entry>::remove(int id) {
-  std::lock_guard<std::mutex> lock(_mutex);
   auto found_iter = _entries.find(id);
   if (found_iter == _entries.end()) {
     throw std::out_of_range("Called IdList::remove() with an invalid ID");
@@ -67,6 +65,11 @@ void IdList<Entry>::remove(int id) {
   _entries.erase(found_iter);
 }
 
+template<class Entry>
+size_t IdList<Entry>::size() const {
+	return _entries.size();
+}
+
 }
 
 #endif

test/fspp/impl/FuseOpenFileListTest.cpp

@@ -45,24 +45,26 @@ struct FuseOpenFileListTest: public ::testing::Test {
     return open(FILEID1, FILEID2);
   }
   void check(int id, int fileid, int flags) {
-    MockOpenFile *openFile = dynamic_cast<MockOpenFile*>(list.get(id));
-    EXPECT_EQ(fileid, openFile->fileid);
-    EXPECT_EQ(flags, openFile->flags);
+	  list.load(id, [=](OpenFile* _openFile) {
+		  MockOpenFile *openFile = dynamic_cast<MockOpenFile*>(_openFile);
+		  EXPECT_EQ(fileid, openFile->fileid);
+		  EXPECT_EQ(flags, openFile->flags);
+	  });
   }
 };
 
 TEST_F(FuseOpenFileListTest, EmptyList1) {
-  ASSERT_THROW(list.get(0), std::out_of_range);
+	ASSERT_THROW(list.load(0, [](OpenFile*) {}), fspp::fuse::FuseErrnoException);
 }
 
 TEST_F(FuseOpenFileListTest, EmptyList2) {
-  ASSERT_THROW(list.get(3), std::out_of_range);
+	ASSERT_THROW(list.load(3, [](OpenFile*) {}), fspp::fuse::FuseErrnoException);
 }
 
 TEST_F(FuseOpenFileListTest, InvalidId) {
   int valid_id = open();
   int invalid_id = valid_id + 1;
-  ASSERT_THROW(list.get(invalid_id), std::out_of_range);
+  ASSERT_THROW(list.load(invalid_id, [](OpenFile*) {}), fspp::fuse::FuseErrnoException);
 }
 
 TEST_F(FuseOpenFileListTest, Open1AndGet) {
@@ -102,18 +104,18 @@ TEST_F(FuseOpenFileListTest, Open3AndGet) {
 TEST_F(FuseOpenFileListTest, GetClosedItemOnEmptyList) {
   int id = open();
 
-  ASSERT_NO_THROW(list.get(id));
+  ASSERT_NO_THROW(list.load(id, [](OpenFile*) {}));
   list.close(id);
-  ASSERT_THROW(list.get(id), std::out_of_range);
+  ASSERT_THROW(list.load(id, [](OpenFile*) {}), fspp::fuse::FuseErrnoException);
 }
 
 TEST_F(FuseOpenFileListTest, GetClosedItemOnNonEmptyList) {
   int id = open();
   open();
 
-  ASSERT_NO_THROW(list.get(id));
+  ASSERT_NO_THROW(list.load(id, [](OpenFile*) {}));
   list.close(id);
-  ASSERT_THROW(list.get(id), std::out_of_range);
+  ASSERT_THROW(list.load(id, [](OpenFile*) {}), fspp::fuse::FuseErrnoException);
 }
 
 TEST_F(FuseOpenFileListTest, CloseOnEmptyList1) {