fstream: add file output stream support

core/fstream.cc

@@ -20,7 +20,9 @@
  */
 
 #include "fstream.hh"
+#include "align.hh"
 #include <malloc.h>
+#include <string.h>
 
 class file_data_source_impl : public data_source_impl {
     lw_shared_ptr<file> _file;
@@ -63,3 +65,56 @@ input_stream<char> make_file_input_stream(
         lw_shared_ptr<file> f, uint64_t offset, size_t buffer_size) {
     return input_stream<char>(file_data_source(std::move(f), offset, buffer_size));
 }
+
+class file_data_sink_impl : public data_sink_impl {
+    lw_shared_ptr<file> _file;
+    size_t _buffer_size;
+    uint64_t _pos = 0;
+public:
+    file_data_sink_impl(lw_shared_ptr<file> f, size_t buffer_size)
+            : _file(std::move(f)), _buffer_size(buffer_size) {}
+    future<> put(net::packet data) { return make_ready_future<>(); }
+    virtual temporary_buffer<char> allocate_buffer(size_t size) override {
+        // buffers to dma_write must be aligned to 512 bytes.
+        return temporary_buffer<char>::aligned(512, size);
+    }
+    virtual future<> put(temporary_buffer<char> buf) override {
+        bool truncate = false;
+        auto pos = _pos;
+        _pos += buf.size();
+        auto p = static_cast<const char*>(buf.get());
+        size_t buf_size = buf.size();
+
+        if ((buf.size() & 511) != 0) {
+            // If buf size isn't aligned, copy its content into a new aligned buf.
+            // This should only happen when the user calls output_stream::flush().
+            auto tmp = allocate_buffer(align_up(buf.size(), 512UL));
+            ::memcpy(tmp.get_write(), buf.get(), buf.size());
+            buf = std::move(tmp);
+            p = buf.get();
+            buf_size = buf.size();
+            truncate = true;
+        }
+        return _file->dma_write(pos, p, buf_size).then(
+            [this, buf = std::move(buf), truncate] (size_t size) {
+            if (truncate) {
+                return _file->truncate(_pos).then([this] {
+                    return _file->flush();
+                });
+            }
+            return make_ready_future<>();
+        });
+    }
+    future<> close() { return make_ready_future<>(); }
+};
+
+class file_data_sink : public data_sink {
+public:
+    file_data_sink(lw_shared_ptr<file> f, size_t buffer_size)
+        : data_sink(std::make_unique<file_data_sink_impl>(
+                std::move(f), buffer_size)) {}
+};
+
+output_stream<char> make_file_output_stream(lw_shared_ptr<file> f, size_t buffer_size) {
+    return output_stream<char>(file_data_sink(std::move(f), buffer_size), buffer_size);
+}

core/fstream.hh

@@ -38,3 +38,10 @@
 input_stream<char> make_file_input_stream(
         lw_shared_ptr<file> file, uint64_t offset = 0,
         uint64_t buffer_size = 8192);
+
+// Create an output_stream for reading starting at the position zero of a
+// newly created file.
+// NOTE: flush() should be the last thing to be called on a file output stream.
+output_stream<char> make_file_output_stream(
+        lw_shared_ptr<file> file,
+        uint64_t buffer_size = 8192);

core/iostream.hh

@@ -144,7 +144,7 @@ private:
 // The data sink will not receive empty chunks.
 //
 template <typename CharType>
-class output_stream {
+class output_stream final {
     static_assert(sizeof(CharType) == 1, "must buffer stream of bytes");
     data_sink _fd;
     temporary_buffer<CharType> _buf;