Merge seastar upstream

core/file.hh

@@ -109,13 +109,13 @@ public:
 };
 
 inline
-std::unique_ptr<file_impl>
+shared_ptr<file_impl>
 make_file_impl(int fd) {
     auto r = ::ioctl(fd, BLKGETSIZE);
     if (r != -1) {
-        return std::unique_ptr<file_impl>(new blockdev_file_impl(fd));
+        return make_shared<blockdev_file_impl>(fd);
     } else {
-        return std::unique_ptr<file_impl>(new posix_file_impl(fd));
+        return make_shared<posix_file_impl>(fd);
     }
 }
 
@@ -131,13 +131,23 @@ make_file_impl(int fd) {
 /// restrictions on file offsets and data pointers.  The former must be aligned
 /// on a 4096 byte boundary, while a 512 byte boundary suffices for the latter.
 class file {
-    std::unique_ptr<file_impl> _file_impl;
+    shared_ptr<file_impl> _file_impl;
 private:
     explicit file(int fd) : _file_impl(make_file_impl(fd)) {}
 
 public:
+    /// Copies a file object.  The new and old objects refer to the
+    /// same underlying file.
+    ///
+    /// \param x file object to be copied
+    file(const file& x) = default;
     /// Moves a file object.
     file(file&& x) : _file_impl(std::move(x._file_impl)) {}
+    /// Assigns a file object.  After assignent, the destination and source refer
+    /// to the same underlying file.
+    ///
+    /// \param x file object to assign to `this`.
+    file& operator=(const file& x) noexcept = default;
     /// Moves assigns a file object.
     file& operator=(file&& x) noexcept = default;
 

core/reactor.cc

@@ -1995,6 +1995,53 @@ future<> touch_directory(sstring name) {
     });
 }
 
+/// \cond internal
+future<> do_flush_directory(sstring name) {
+    if (name.empty()) {
+        return make_ready_future<>();
+    }
+
+    return open_directory(name).then([] (file f) {
+        auto fptr = std::make_unique<file>(std::move(f));
+        auto fut = fptr->flush();
+        return fut.then([fptr = std::move(fptr)] {
+            return fptr->close();
+        });
+    });
+}
+
+future<> do_recursive_touch_directory(sstring base, sstring name) {
+    static const sstring::value_type separator = '/';
+
+    if (name.empty()) {
+        return make_ready_future<>();
+    }
+
+    size_t pos = std::min(name.find(separator), name.size() - 1);
+    base += name.substr(0 , pos + 1);
+    name = name.substr(pos + 1);
+    return touch_directory(base).then([base, name] {
+        return do_recursive_touch_directory(base, name);
+    }).then([base] {
+        // We will now flush the directory that holds the entry we potentially
+        // created. Technically speaking, we only need to touch when we did
+        // create. But flushing the unchanged ones should be cheap enough - and
+        // it simplifies the code considerably.
+        return do_flush_directory(base);
+    });
+}
+/// \endcond
+
+future<> recursive_touch_directory(sstring name) {
+    // If the name is empty,  it will be of the type a/b/c, which should be interpreted as
+    // a relative path. This means we have to flush our current directory
+    sstring base = "";
+    if (name[0] == '/' || name[0] == '.') {
+        base = "./";
+    }
+    return do_recursive_touch_directory(base, name);
+}
+
 future<> remove_file(sstring pathname) {
     return engine().remove_file(std::move(pathname));
 }

core/seastar.hh

@@ -169,6 +169,17 @@ future<> make_directory(sstring name);
 /// containing directory is sync'ed.
 future<> touch_directory(sstring name);
 
+/// Recursively ensures a directory exists
+///
+/// Checks whether each component of a directory exists, and if not, creates it.
+///
+/// \param name name of the directory to potentially create
+/// \param separator character used as directory separator
+///
+/// \note
+/// This function fsyncs each component created, and is therefore guaranteed to be stable on disk.
+future<> recursive_touch_directory(sstring name);
+
 /// Removes (unlinks) a file.
 ///
 /// \param name name of the file to remove