Skip to content

ARROW-10012: [C++] Make MockFileSystem thread-safe #8193

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
pitrou wants to merge 1 commit into from
+41 −1
Closed

ARROW-10012: [C++] Make MockFileSystem thread-safe #8193

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 41 additions & 1 deletion cpp/src/arrow/filesystem/mockfs.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
#include <algorithm>
#include <iterator>
#include <map>
#include <mutex>
#include <sstream>
#include <string>
#include <utility>
Expand Down Expand Up @@ -224,10 +225,15 @@ class MockFileSystem::Impl {
TimePoint current_time;
// The root directory
Entry root;
std::mutex mutex;

explicit Impl(TimePoint current_time)
: current_time(current_time), root(Directory("", current_time)) {}

std::unique_lock<std::mutex> lock_guard() {
return std::unique_lock<std::mutex>(mutex);
}

Directory& RootDir() { return root.as_dir(); }

template <typename It>
Expand Down Expand Up @@ -376,12 +382,14 @@ Status MockFileSystem::CreateDir(const std::string& path, bool recursive) {
auto parts = SplitAbstractPath(path);
RETURN_NOT_OK(ValidateAbstractPathParts(parts));

auto guard = impl_->lock_guard();

size_t consumed;
Entry* entry = impl_->FindEntry(parts, &consumed);
if (!entry->is_dir()) {
auto file_path = JoinAbstractPath(parts.begin(), parts.begin() + consumed);
return Status::IOError("Cannot create directory '", path, "': ", "ancestor '",
file_path, "' is a regular file");
file_path, "' is not a directory");
}
if (!recursive && (parts.size() - consumed) > 1) {
return Status::IOError("Cannot create directory '", path,
Expand All @@ -392,6 +400,7 @@ Status MockFileSystem::CreateDir(const std::string& path, bool recursive) {
std::unique_ptr<Entry> child(new Entry(Directory(name, impl_->current_time)));
Entry* child_ptr = child.get();
bool inserted = entry->as_dir().CreateEntry(name, std::move(child));
// No race condition on insertion is possible, as all operations are locked
DCHECK(inserted);
entry = child_ptr;
}
Expand All @@ -402,6 +411,8 @@ Status MockFileSystem::DeleteDir(const std::string& path) {
auto parts = SplitAbstractPath(path);
RETURN_NOT_OK(ValidateAbstractPathParts(parts));

auto guard = impl_->lock_guard();

Entry* parent = impl_->FindParent(parts);
if (parent == nullptr || !parent->is_dir()) {
return PathNotFound(path);
Expand All @@ -424,6 +435,8 @@ Status MockFileSystem::DeleteDirContents(const std::string& path) {
auto parts = SplitAbstractPath(path);
RETURN_NOT_OK(ValidateAbstractPathParts(parts));

auto guard = impl_->lock_guard();

if (parts.empty()) {
// Wipe filesystem
return internal::InvalidDeleteDirContents(path);
Expand All @@ -441,6 +454,8 @@ Status MockFileSystem::DeleteDirContents(const std::string& path) {
}

Status MockFileSystem::DeleteRootDirContents() {
auto guard = impl_->lock_guard();

impl_->RootDir().entries.clear();
return Status::OK();
}
Expand All @@ -449,6 +464,8 @@ Status MockFileSystem::DeleteFile(const std::string& path) {
auto parts = SplitAbstractPath(path);
RETURN_NOT_OK(ValidateAbstractPathParts(parts));

auto guard = impl_->lock_guard();

Entry* parent = impl_->FindParent(parts);
if (parent == nullptr || !parent->is_dir()) {
return PathNotFound(path);
Expand All @@ -470,6 +487,8 @@ Result<FileInfo> MockFileSystem::GetFileInfo(const std::string& path) {
auto parts = SplitAbstractPath(path);
RETURN_NOT_OK(ValidateAbstractPathParts(parts));

auto guard = impl_->lock_guard();

FileInfo info;
Entry* entry = impl_->FindEntry(parts);
if (entry == nullptr) {
Expand All @@ -485,6 +504,8 @@ Result<std::vector<FileInfo>> MockFileSystem::GetFileInfo(const FileSelector& se
auto parts = SplitAbstractPath(selector.base_dir);
RETURN_NOT_OK(ValidateAbstractPathParts(parts));

auto guard = impl_->lock_guard();

std::vector<FileInfo> results;

Entry* base_dir = impl_->FindEntry(parts);
Expand All @@ -504,6 +525,8 @@ Result<std::vector<FileInfo>> MockFileSystem::GetFileInfo(const FileSelector& se
return results;
}

namespace {

// Helper for binary operations (move, copy)
struct BinaryOp {
std::vector<std::string> src_parts;
Expand All @@ -523,6 +546,8 @@ struct BinaryOp {
RETURN_NOT_OK(ValidateAbstractPathParts(src_parts));
RETURN_NOT_OK(ValidateAbstractPathParts(dest_parts));

auto guard = impl->lock_guard();

// Both source and destination must have valid parents
Entry* src_parent = impl->FindParent(src_parts);
if (src_parent == nullptr || !src_parent->is_dir()) {
Expand Down Expand Up @@ -552,6 +577,8 @@ struct BinaryOp {
}
};

} // namespace

Status MockFileSystem::Move(const std::string& src, const std::string& dest) {
return BinaryOp::Run(impl_.get(), src, dest, [&](const BinaryOp& op) -> Status {
if (op.src_entry == nullptr) {
Expand Down Expand Up @@ -609,31 +636,43 @@ Status MockFileSystem::CopyFile(const std::string& src, const std::string& dest)

Result<std::shared_ptr<io::InputStream>> MockFileSystem::OpenInputStream(
const std::string& path) {
auto guard = impl_->lock_guard();

return impl_->OpenInputReader(path);
}

Result<std::shared_ptr<io::RandomAccessFile>> MockFileSystem::OpenInputFile(
const std::string& path) {
auto guard = impl_->lock_guard();

return impl_->OpenInputReader(path);
}

Result<std::shared_ptr<io::OutputStream>> MockFileSystem::OpenOutputStream(
const std::string& path) {
auto guard = impl_->lock_guard();

return impl_->OpenOutputStream(path, false /* append */);
}

Result<std::shared_ptr<io::OutputStream>> MockFileSystem::OpenAppendStream(
const std::string& path) {
auto guard = impl_->lock_guard();

return impl_->OpenOutputStream(path, true /* append */);
}

std::vector<MockDirInfo> MockFileSystem::AllDirs() {
auto guard = impl_->lock_guard();

std::vector<MockDirInfo> result;
impl_->DumpDirs("", impl_->RootDir(), &result);
return result;
}

std::vector<MockFileInfo> MockFileSystem::AllFiles() {
auto guard = impl_->lock_guard();

std::vector<MockFileInfo> result;
impl_->DumpFiles("", impl_->RootDir(), &result);
return result;
Expand All @@ -642,6 +681,7 @@ std::vector<MockFileInfo> MockFileSystem::AllFiles() {
Status MockFileSystem::CreateFile(const std::string& path, const std::string& contents,
bool recursive) {
auto parent = fs::internal::GetAbstractPathParent(path).first;

if (parent != "") {
RETURN_NOT_OK(CreateDir(parent, recursive));
}
Expand Down