util/file.hh: implement remove_file_via_blocks_discarding()

In order to allow splitting the removal of a big file into
smaller steps a new function has been implemented. It removes
a regular file via blocks discarding -- step by step.

The discard requests are processed by the I/O queue. Therefore,
they are issued to the disk according to the given configuration.
The removal may be delayed when the bandwidth is not available in
the queue.

Fundamentally, the new function performs four steps:
 1. Open the file.
 2. Unlink it.
 3. Punch holes until nothing is left.
 4. Close the file.

Signed-off-by: Patryk Wrobel <[email protected]>

include/seastar/util/file.hh

@@ -102,6 +102,18 @@ future<std::vector<temporary_buffer<char>>> read_entire_file(std::filesystem::pa
 /// \param path path of the file to be read.
 future<sstring> read_entire_file_contiguous(std::filesystem::path path);
 
+/// Removes a regular file via blocks discarding.
+///
+/// \param name name of the file to remove.
+///
+/// \note
+/// The removal is processed by the I/O queue according to its configuration.
+/// The removal may be delayed when the bandwidth is not available. The user
+/// must ensure, that the file is not used until the removal finishes.
+///
+/// Effectively: opens a file, unlinks it and punches holes until nothing is left.
+future<> remove_file_via_blocks_discarding(std::string_view name) noexcept;
+
 SEASTAR_MODULE_EXPORT_END
 /// @}
 

src/core/file.cc

@@ -268,7 +268,11 @@ posix_file_impl::fcntl_short(int op, uintptr_t arg) noexcept {
 
 future<>
 posix_file_impl::discard(uint64_t offset, uint64_t length) noexcept {
-    return engine().punch_hole(_fd, offset, length);
+    internal::maybe_priority_class_ref io_priority_class;
+    auto req = internal::io_request::make_discard(_fd, offset, length);
+    return _io_queue.submit_io_discard(internal::priority_class(io_priority_class), length, std::move(req), nullptr).then([] (auto len) {
+        return make_ready_future<>();
+    });
 }
 
 future<>

src/util/file.cc

@@ -37,6 +37,7 @@ module seastar;
 #else
 #include <seastar/core/reactor.hh>
 #include <seastar/core/seastar.hh>
+#include <seastar/core/when_all.hh>
 #include <seastar/util/file.hh>
 #endif
 
@@ -227,6 +228,70 @@ future<sstring> read_entire_file_contiguous(std::filesystem::path path) {
     });
 }
 
+future<> remove_file_via_blocks_discarding(std::string_view name) noexcept {
+    return open_file_dma(name, open_flags::rw, file_open_options{}).then([name = sstring(name)](file f) mutable {
+        return f.size().then_wrapped([f, name = std::move(name)](future<uint64_t> fut) mutable {
+            // If getting size failed then close file and report error gracefully.
+            if (fut.failed()) {
+                return f.close().then([f, ex = fut.get_exception()] () mutable {
+                    return make_exception_future(ex);
+                });
+            }
+
+            return remove_file(name).then_wrapped([f, fsize = fut.get()] (future<> fut) mutable {
+                // If unlink failed, then close file and report error gracefully.
+                if (fut.failed()) {
+                    return f.close().then([f, ex = fut.get_exception()] () mutable {
+                        return make_exception_future(ex);
+                    });
+                }
+
+                // If file was empty, then unlink is sufficient.
+                if (fsize == 0u) {
+                    return f.close().finally([f](){});
+                }
+
+                // Else, issue discards for all blocks.
+                const uint64_t block_size{32u << 20u}; // 32MiB
+                const uint64_t additional_iteration = (fsize % block_size == 0) ? 0 : 1;
+                const uint64_t blocks_count{static_cast<uint64_t>(fsize / block_size) + additional_iteration};
+
+                lw_shared_ptr<std::vector<future<>>> futs;
+                try {
+                    futs = make_lw_shared<std::vector<future<>>>();
+                    futs->reserve(blocks_count);
+                } catch (...) {
+                    return f.close().then([f, ex = std::current_exception()] {
+                        return make_exception_future(ex);
+                    });
+                }
+
+                for (uint64_t i = 0u; i < blocks_count; ++i) {
+                    auto offset = i * block_size;
+                    auto discard_end = std::min(offset + block_size, fsize);
+                    auto length = discard_end - offset;
+                    futs->push_back(f.discard(offset, length));
+                }
+
+                // Wait until all finish and close file.
+                return when_all(futs->begin(), futs->end()).then([futs, f] (auto results) mutable {
+                    for (auto&& res : results) {
+                        if (res.failed()) {
+                            return f.close().then([f, ex = res.get_exception()] () mutable {
+                                return make_exception_future<>(ex);
+                            });
+                        }
+
+                        res.ignore_ready_future();
+                    }
+
+                    return f.close().finally([f](){});
+                });
+            });
+        });
+    });
+}
+
 } // namespace util
 
 } //namespace seastar