Add FileSystemBuffer::Fragment class to support file_system_buffer

CODEOWNERS

@@ -187,8 +187,9 @@ extensions/filters/http/oauth2 @rgs1 @derekargueta @snowp
 /*/extensions/key_value @alyssawilk @ryantheoptimist
 # Config Validators
 /*/extensions/config/validators/minimum_clusters @adisuissa @htuch
-# support library for filesystem based extensions
+# File system based extensions
 /*/extensions/common/async_files @mattklein123 @ravenblack
+/*/extensions/filters/http/file_system_buffer @mattklein123 @ravenblack
 # Google Cloud Platform Authentication Filter
 /*/extensions/filters/http/gcp_authn @tyxia @yanavlasov
 # DNS resolution

source/extensions/filters/http/file_system_buffer/BUILD

@@ -0,0 +1,18 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_library(
+    name = "fragment",
+    srcs = ["fragment.cc"],
+    hdrs = ["fragment.h"],
+    deps = [
+        "//source/extensions/common/async_files",
+    ],
+)

source/extensions/filters/http/file_system_buffer/fragment.cc

@@ -0,0 +1,134 @@
+#include "source/extensions/filters/http/file_system_buffer/fragment.h"
+
+#include "source/common/buffer/buffer_impl.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace FileSystemBuffer {
+
+class FragmentData {
+public:
+  virtual bool isMemory() const { return false; }
+  virtual bool isStorage() const { return false; }
+  virtual ~FragmentData() = default;
+};
+
+class MemoryFragment : public FragmentData {
+public:
+  explicit MemoryFragment(Buffer::Instance& buffer);
+  explicit MemoryFragment(Buffer::Instance& buffer, size_t size);
+  std::unique_ptr<Buffer::Instance> extract();
+  bool isMemory() const override { return true; }
+
+private:
+  std::unique_ptr<Buffer::OwnedImpl> buffer_;
+};
+
+class WritingFragment : public FragmentData {};
+
+class ReadingFragment : public FragmentData {};
+
+class StorageFragment : public FragmentData {
+public:
+  explicit StorageFragment(off_t offset) : offset_(offset) {}
+  off_t offset() const { return offset_; }
+  bool isStorage() const override { return true; }
+
+private:
+  const off_t offset_;
+};
+
+Fragment::Fragment(Buffer::Instance& buffer)
+    : size_(buffer.length()), data_(std::make_unique<MemoryFragment>(buffer)) {}
+
+Fragment::Fragment(Buffer::Instance& buffer, size_t size)
+    : size_(size), data_(std::make_unique<MemoryFragment>(buffer, size)) {}
+
+Fragment::~Fragment() = default;
+bool Fragment::isMemory() const { return data_->isMemory(); }
+bool Fragment::isStorage() const { return data_->isStorage(); }
+
+MemoryFragment::MemoryFragment(Buffer::Instance& buffer)
+    : buffer_(std::make_unique<Buffer::OwnedImpl>()) {
+  buffer_->move(buffer);
+}
+
+MemoryFragment::MemoryFragment(Buffer::Instance& buffer, size_t size)
+    : buffer_(std::make_unique<Buffer::OwnedImpl>()) {
+  buffer_->move(buffer, size);
+}
+
+std::unique_ptr<Buffer::Instance> Fragment::extract() {
+  ASSERT(isMemory());
+  auto ret = dynamic_cast<MemoryFragment*>(data_.get())->extract();
+  data_.reset();
+  size_ = 0;
+  return ret;
+}
+
+std::unique_ptr<Buffer::Instance> MemoryFragment::extract() { return std::move(buffer_); }
+
+absl::StatusOr<CancelFunction>
+Fragment::toStorage(AsyncFileHandle file, off_t offset,
+                    std::function<void(absl::StatusOr<UpdateFragmentFunction>)> on_done) {
+  ASSERT(isMemory());
+  auto data = dynamic_cast<MemoryFragment*>(data_.get())->extract();
+  data_ = std::make_unique<WritingFragment>();
+  return file->write(*data, offset,
+                     [fragment = this, size = size_, offset,
+                      on_done = std::move(on_done)](absl::StatusOr<size_t> result) {
+                       // This lambda always runs in the async file thread, so must not use the
+                       // fragment directly as it may have been destroyed. It is passed through to
+                       // call from on_done's callback, where it can be used again as that function
+                       // is called from the envoy thread only if the filter has *not* been
+                       // destroyed in the meantime.
+                       if (!result.ok()) {
+                         on_done(result.status());
+                         return;
+                       }
+                       if (result.value() != size) {
+                         on_done(absl::AbortedError(fmt::format(
+                             "buffer write wrote {} bytes, wanted {}", result.value(), size)));
+                         return;
+                       }
+                       on_done([fragment, offset]() {
+                         fragment->data_ = std::make_unique<StorageFragment>(offset);
+                       });
+                     });
+}
+
+absl::StatusOr<CancelFunction>
+Fragment::fromStorage(AsyncFileHandle file,
+                      std::function<void(absl::StatusOr<UpdateFragmentFunction>)> on_done) {
+  ASSERT(isStorage());
+  off_t offset = dynamic_cast<StorageFragment*>(data_.get())->offset();
+  data_ = std::make_unique<ReadingFragment>();
+  return file->read(
+      offset, size_,
+      [fragment = this, size = size_,
+       on_done](absl::StatusOr<std::unique_ptr<Buffer::Instance>> result) {
+        // This lambda always runs in the async file thread, so must not use the
+        // fragment directly as it may have been destroyed. It is passed through to
+        // call from on_done's callback, where it can be used again as that function
+        // is called from the envoy thread only if the filter has *not* been destroyed
+        // in the meantime.
+        if (!result.ok()) {
+          on_done(result.status());
+          return;
+        }
+        if (result.value()->length() != size) {
+          on_done(absl::AbortedError(
+              fmt::format("buffer read got {} bytes, wanted {}", result.value()->length(), size)));
+          return;
+        }
+        on_done([fragment, data = std::shared_ptr<Buffer::Instance>(result.value().release())]() {
+          fragment->data_ = std::make_unique<MemoryFragment>(*data);
+        });
+      });
+}
+
+} // namespace FileSystemBuffer
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

source/extensions/filters/http/file_system_buffer/fragment.h

@@ -0,0 +1,102 @@
+#pragma once
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "envoy/buffer/buffer.h"
+
+#include "source/extensions/common/async_files/async_file_handle.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace HttpFilters {
+namespace FileSystemBuffer {
+
+class FragmentData;
+using UpdateFragmentFunction = std::function<void()>;
+using Extensions::Common::AsyncFiles::AsyncFileHandle;
+using Extensions::Common::AsyncFiles::CancelFunction;
+
+// A Fragment is a piece of the buffer queue used by the filter.
+//
+// Each fragment may be in memory, on disk, or in an unusable intermediate state
+// while controlled by the AsyncFiles library.
+//
+// The fragments are queued in sequential order, and may not be in the same order
+// in the buffer file, as we write "last fragment first" so that the soonest-needed
+// fragments remain in memory as long as possible.
+//
+// Memory fragments are simply `Buffer::Instance`s. Storage fragments contain just
+// enough information to reload a memory fragment from the buffer file.
+class Fragment {
+public:
+  explicit Fragment(Buffer::Instance& buffer);
+  explicit Fragment(Buffer::Instance& buffer, size_t size);
+  ~Fragment();
+
+  // Starts the transition for this fragment from memory to storage.
+  //
+  // The on_done callback is called when the file write completes.
+  //
+  // The callback includes its own callback parameter update_fragment (on success),
+  // so that the operation of updating the fragment can be dispatched to be performed
+  // in the envoy worker thread, rather than being performed on the callback thread.
+  // (This is important to allow for proper handling if the context was deleted while
+  // the operation was in flight, for example.)
+  // i.e. typical usage resembles:
+  // auto result = fragment.toStorage(file_handle, offset,
+  //     [this](absl::StatusOr<UpdateFragmentFunction> status_or_update_fragment) {
+  //   if (status_or_update_fragment.ok()) {
+  //     auto update_fragment = status_or_update_fragment.value();
+  //     dispatcher->dispatch([this, update_fragment]() {
+  //       update_fragment();
+  //       // do more stuff with the fragment, or in reaction to it being updated.
+  //     });
+  //   } else {
+  //     dispatcher->dispatch([this, status = status_or_update_fragment.status()]() {
+  //       // do stuff in response to the error.
+  //     });
+  //   }
+  // });
+  absl::StatusOr<CancelFunction>
+  toStorage(AsyncFileHandle file, off_t offset,
+            std::function<void(absl::StatusOr<UpdateFragmentFunction>)> on_done);
+
+  // Starts the transition for this fragment from storage to memory.
+  //
+  // The on_done callback is called when the file read completes.
+  //
+  // The callback includes its own callback parameter update_fragment (on success),
+  // so that the operation of updating the fragment can be dispatched to be performed
+  // in the envoy worker thread, rather than being performed on the callback thread.
+  // (This is important to allow for proper handling if the context was deleted while
+  // the operation was in flight, for example.)
+  //
+  // See toStorage for example usage.
+  absl::StatusOr<CancelFunction>
+  fromStorage(AsyncFileHandle file,
+              std::function<void(absl::StatusOr<UpdateFragmentFunction>)> on_done);
+
+  // Removes the buffer from a memory instance and resets it to size 0.
+  //
+  // It is an error to call extract() on a Fragment that is not in memory.
+  Buffer::InstancePtr extract();
+
+  // Returns true if the fragment is in memory, false if in storage or transition.
+  bool isMemory() const;
+
+  // Returns true if the fragment is in storage, false if in memory or transition.
+  bool isStorage() const;
+
+  // Returns the size of the fragment in bytes.
+  size_t size() const { return size_; }
+
+private:
+  size_t size_;
+  std::unique_ptr<FragmentData> data_;
+};
+
+} // namespace FileSystemBuffer
+} // namespace HttpFilters
+} // namespace Extensions
+} // namespace Envoy

test/extensions/filters/http/file_system_buffer/BUILD

@@ -0,0 +1,21 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_test",
+    "envoy_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_package()
+
+envoy_cc_test(
+    name = "fragment_test",
+    srcs = ["fragment_test.cc"],
+    tags = ["skip_on_windows"],
+    deps = [
+        "//source/extensions/filters/http/file_system_buffer:fragment",
+        "//test/extensions/common/async_files:mocks",
+        "//test/mocks/buffer:buffer_mocks",
+        "//test/test_common:status_utility_lib",
+    ],
+)