Add async_files library to support filesystem buffering and filesystem caching

CODEOWNERS

@@ -187,6 +187,8 @@ 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
+/*/extensions/common/async_files @mattklein123 @ravenblack
 # Google Cloud Platform Authentication Filter
 /*/extensions/filters/http/gcp_authn @tyxia @yanavlasov
 # DNS resolution

api/BUILD

@@ -126,6 +126,7 @@ proto_library(
         "//envoy/extensions/clusters/aggregate/v3:pkg",
         "//envoy/extensions/clusters/dynamic_forward_proxy/v3:pkg",
         "//envoy/extensions/clusters/redis/v3:pkg",
+        "//envoy/extensions/common/async_files/v3:pkg",
         "//envoy/extensions/common/dynamic_forward_proxy/v3:pkg",
         "//envoy/extensions/common/matching/v3:pkg",
         "//envoy/extensions/common/ratelimit/v3:pkg",

api/envoy/extensions/common/async_files/v3/BUILD

@@ -0,0 +1,12 @@
+# DO NOT EDIT. This file is generated by tools/proto_format/proto_sync.py.
+
+load("@envoy_api//bazel:api_build_system.bzl", "api_proto_package")
+
+licenses(["notice"])  # Apache 2
+
+api_proto_package(
+    deps = [
+        "@com_github_cncf_udpa//udpa/annotations:pkg",
+        "@com_github_cncf_udpa//xds/annotations/v3:pkg",
+    ],
+)

api/envoy/extensions/common/async_files/v3/async_file_manager.proto

@@ -0,0 +1,41 @@
+syntax = "proto3";
+
+package envoy.extensions.common.async_files.v3;
+
+import "xds/annotations/v3/status.proto";
+
+import "udpa/annotations/status.proto";
+import "validate/validate.proto";
+
+option java_package = "io.envoyproxy.envoy.extensions.common.async_files.v3";
+option java_outer_classname = "AsyncFileManagerProto";
+option java_multiple_files = true;
+option go_package = "github.com/envoyproxy/go-control-plane/envoy/extensions/common/async_files/v3;async_filesv3";
+option (udpa.annotations.file_status).package_version_status = ACTIVE;
+option (xds.annotations.v3.file_status).work_in_progress = true;
+
+// [#protodoc-title: AsyncFileManager configuration]
+
+// Configuration to instantiate or select a singleton `AsyncFileManager`.
+message AsyncFileManagerConfig {
+  message ThreadPool {
+    // The number of threads to use. If unset or zero, will default to the number
+    // of concurrent threads the hardware supports. This default is subject to
+    // change if performance analysis suggests it.
+    uint32 thread_count = 1;
+  }
+
+  // An optional identifier for the manager. An empty string is a valid identifier
+  // for a common, default `AsyncFileManager`.
+  //
+  // Reusing the same id with different configurations in the same envoy instance
+  // is an error.
+  string id = 1;
+
+  oneof manager_type {
+    option (validate.required) = true;
+
+    // Configuration for a thread-pool based async file manager.
+    ThreadPool thread_pool = 2;
+  }
+}

api/versioning/BUILD

@@ -63,6 +63,7 @@ proto_library(
         "//envoy/extensions/clusters/aggregate/v3:pkg",
         "//envoy/extensions/clusters/dynamic_forward_proxy/v3:pkg",
         "//envoy/extensions/clusters/redis/v3:pkg",
+        "//envoy/extensions/common/async_files/v3:pkg",
         "//envoy/extensions/common/dynamic_forward_proxy/v3:pkg",
         "//envoy/extensions/common/matching/v3:pkg",
         "//envoy/extensions/common/ratelimit/v3:pkg",

docs/root/api-v3/common_messages/common_messages.rst

@@ -26,6 +26,7 @@ Common messages
   ../extensions/filters/common/fault/v3/fault.proto
   ../extensions/network/socket_interface/v3/default_socket_interface.proto
   ../extensions/common/matching/v3/extension_matcher.proto
+  ../extensions/common/async_files/v3/async_file_manager.proto
   ../extensions/filters/common/dependency/v3/dependency.proto
   ../extensions/filters/common/matcher/action/v3/skip_action.proto
   ../extensions/key_value/file_based/v3/config.proto

source/extensions/common/async_files/BUILD

@@ -0,0 +1,83 @@
+load(
+    "//bazel:envoy_build_system.bzl",
+    "envoy_cc_library",
+    "envoy_extension_package",
+)
+
+licenses(["notice"])  # Apache 2
+
+envoy_extension_package()
+
+envoy_cc_library(
+    name = "async_files_base",
+    srcs = [
+        "async_file_action.cc",
+        "async_file_context_base.cc",
+    ],
+    hdrs = [
+        "async_file_action.h",
+        "async_file_context_base.h",
+        "async_file_handle.h",
+        "async_file_manager.h",
+    ],
+    deps = [
+        ":status_after_file_error",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/common:utility_lib",
+        "@com_google_absl//absl/base",
+        "@com_google_absl//absl/status:statusor",
+    ],
+)
+
+envoy_cc_library(
+    name = "async_files_thread_pool",
+    srcs = [
+        "async_file_context_thread_pool.cc",
+        "async_file_manager_thread_pool.cc",
+    ],
+    hdrs = [
+        "async_file_context_thread_pool.h",
+        "async_file_manager_thread_pool.h",
+    ],
+    deps = [
+        ":async_files_base",
+        ":status_after_file_error",
+        "//source/common/api:os_sys_calls_lib",
+        "//source/common/buffer:buffer_lib",
+        "@com_google_absl//absl/base",
+        "@com_google_absl//absl/status:statusor",
+        "@envoy_api//envoy/extensions/common/async_files/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_library(
+    name = "async_files",
+    srcs = [
+        "async_file_manager.cc",
+        "async_file_manager_factory.cc",
+    ],
+    hdrs = [
+        "async_file_manager_factory.h",
+    ],
+    deps = [
+        ":async_files_thread_pool",
+        "//source/common/api:os_sys_calls_lib",
+        "//source/common/buffer:buffer_lib",
+        "//source/common/protobuf:utility_lib",
+        "@com_google_absl//absl/base",
+        "@com_google_absl//absl/status:statusor",
+        "@envoy_api//envoy/extensions/common/async_files/v3:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_library(
+    name = "status_after_file_error",
+    srcs = ["status_after_file_error.cc"],
+    hdrs = ["status_after_file_error.h"],
+    deps = [
+        "//envoy/api:os_sys_calls_interface",
+        "//source/common/common:assert_lib",
+        "//source/common/common:utility_lib",
+        "@com_google_absl//absl/status",
+    ],
+)

source/extensions/common/async_files/README.md

@@ -0,0 +1,65 @@
+# AsyncFileManager
+
+An `AsyncFileManager` should be a singleton or similarly long-lived scope. It represents a
+thread pool for performing file operations asynchronously.
+
+`AsyncFileManager` can create `AsyncFileHandle`s via `createAnonymousFile` or `openExistingFile`,
+can postpone queuing file actions using `whenReady`, and can delete files via `unlink`.
+
+# AsyncFileHandle
+
+An `AsyncFileHandle` represents a context in which asynchronous file operations can be performed. It is associated with at most one file at a time.
+
+Each action on an AsyncFileHandle is effectively an "enqueue" action, in that it places the action in the manager's execution queue, it does not immediately perform the requested action. Actions on an `AsyncFileHandle` can be *chained*, by enqueuing another action during the callback from a previous action, e.g.
+
+```
+manager->createAnonymousFile("/tmp", [](absl::StatusOr<AsyncFileHandle> opened) {
+  if (!opened.ok()) {
+    std::cout << "oh no, an error: " << opened.status() << std::endl;
+    return;
+  }
+  auto handle = opened.value();
+  handle->write(someBuffer, 0, [handle](absl::StatusOr<size_t> written) {
+    if (!written.ok()) {
+      std::cout << "oh no, an error: " << written.status() << std::endl;
+      return;
+    }
+    std::cout << "wrote " << written.value() << " bytes" << std::endl;
+    handle->close([](absl::Status closed) {
+      if (!closed.ok()) {
+        std::cout << "oh no, an error: " << closed << std::endl;
+      }
+    }).IgnoreError(); // A returned error only occurs if the file handle was closed.
+  }).IgnoreError(); // A returned error only occurs if the file handle was closed.
+});
+```
+
+Will open an unnamed file, write 5 bytes, and close it. (This is just for explanatory purposes, in practice you would most likely want the callbacks to call something on `this` rather than nesting lambdas!)
+
+Chaining actions, as opposed to enqueuing, passing the result to a main thread, and from there enqueuing again, will not yield the thread in a thread-pool based implementation. An advantage of this is that, for example, if 5 workers all wanted to write a 100kb file at the same moment, with unchained requests in a one-thread threadpool the sequence would most likely resemble
+
+```
+OPEN-OPEN-OPEN-OPEN-OPEN-WRITE-WRITE-WRITE-WRITE-WRITE-CLOSE-CLOSE-CLOSE-CLOSE-CLOSE
+```
+
+Versus with appropriately chained requests in a one-thread threadpool the sequence would be guaranteed to be
+
+```
+OPEN-WRITE-CLOSE-OPEN-WRITE-CLOSE-OPEN-WRITE-CLOSE-OPEN-WRITE-CLOSE-OPEN-WRITE-CLOSE
+```
+
+Expand this concept to 100+ files all asking to be written at once and you can immediately see the advantages of chaining; not having the resource issues of many files open at the same time, more localized access, etc.
+
+## cancellation
+
+Each action function returns a cancellation function which can be called to remove an action from the queue and prevent the callback from being called. If the execution is already in progress, it may be undone (e.g. a file open operation will close the file if it is opening when cancel is called). The cancel function will block if the callback is already in progress when cancel is called, until the callback completes. This should not be a long block, as callbacks should be short (see callbacks below).
+
+As such, a client should ensure that the cleanup order is consistent - if a callback captures a file handle, the client should clean up that file handle (if present) *after* calling cancel, in case the file was opened during the call to cancel.
+
+## callbacks
+
+The callbacks passed to `AsyncFileHandle` and `AsyncFileManager` are scheduled in a thread or thread pool belonging to the AsyncFileManager - therefore they should be doing minimal work, not blocking (for more than a trivial data-guard lock), and return promptly. If any significant work or blocking is required, the result of the previous action should be passed from the callback to another thread (via some dispatcher or other queuing mechanism) so the manager's thread can continue performing file operations for other clients.
+
+## Possible actions
+
+See `async_file_handle.h` for the actions that can currently be queued on an `AsyncFileHandle`.

source/extensions/common/async_files/async_file_action.cc

@@ -0,0 +1,26 @@
+#include "source/extensions/common/async_files/async_file_action.h"
+
+#include <thread>
+
+namespace Envoy {
+namespace Extensions {
+namespace Common {
+namespace AsyncFiles {
+
+void AsyncFileAction::cancel() {
+  auto previousState = state_.exchange(State::Cancelled);
+  if (previousState == State::InCallback) {
+    // A gentle spin-lock. This situation should be rare, and callbacks are
+    // supposed to be quick, so we don't need a real lock here.
+    while (state_.load() != State::Done) {
+      std::this_thread::yield();
+    }
+  } else if (previousState == State::Done) {
+    state_.store(State::Done);
+  }
+}
+
+} // namespace AsyncFiles
+} // namespace Common
+} // namespace Extensions
+} // namespace Envoy

source/extensions/common/async_files/async_file_action.h

@@ -0,0 +1,99 @@
+#pragma once
+
+#include <atomic>
+#include <functional>
+
+#include "envoy/common/pure.h"
+
+#include "source/common/common/assert.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Common {
+namespace AsyncFiles {
+
+// A CancelFunction attempts to stop an action in flight.
+// * If the action already occurred, the CancelFunction does nothing.
+// * If the action is already calling the callback, CancelFunction blocks until the callback
+//   completes.
+// * If the action is already executing, CancelFunction causes the removal of any resource-consuming
+//   return value (e.g. file handles), and prevents the callback.
+// * If the action is still just queued, CancelFunction prevents its execution.
+using CancelFunction = std::function<void()>;
+
+// Actions to be passed to asyncFileManager->enqueue.
+class AsyncFileAction {
+public:
+  virtual ~AsyncFileAction() = default;
+
+  // Cancel the action, as much as possible.
+  //
+  // If the action has not been started, it will become a no-op.
+  //
+  // If the action has started, onCancelledBeforeCallback will be called,
+  // and the callback will not.
+  //
+  // If the callback is already being called, cancel will block until the
+  // callback has completed.
+  //
+  // If the action is already complete, cancel does nothing.
+  void cancel();
+
+  // Performs the action represented by this instance, and calls the callback
+  // on completion or on error.
+  virtual void execute() PURE;
+
+protected:
+  enum class State { Queued, Executing, InCallback, Done, Cancelled };
+  std::atomic<State> state_{State::Queued};
+};
+
+// All concrete AsyncFileActions are a subclass of AsyncFileActionWithResult.
+// The template allows for different on_complete callback signatures appropriate
+// to each specific action.
+//
+// on_complete callbacks run in the AsyncFileManager's thread pool, and therefore:
+// 1. Should avoid using variables that may be out of scope by the time the callback is called.
+// 2. May need to lock-guard variables that can be changed in other threads.
+// 3. Must not block significantly or do significant work - if anything time-consuming is required
+// the result should be passed to another thread for handling.
+template <typename T> class AsyncFileActionWithResult : public AsyncFileAction {
+public:
+  explicit AsyncFileActionWithResult(std::function<void(T)> on_complete)
+      : on_complete_(on_complete) {}
+
+  void execute() final {
+    State expected = State::Queued;
+    if (!state_.compare_exchange_strong(expected, State::Executing)) {
+      ASSERT(expected == State::Cancelled);
+      return;
+    }
+    expected = State::Executing;
+    T result = executeImpl();
+    if (!state_.compare_exchange_strong(expected, State::InCallback)) {
+      ASSERT(expected == State::Cancelled);
+      onCancelledBeforeCallback(std::move(result));
+      return;
+    }
+    on_complete_(std::move(result));
+    state_.store(State::Done);
+  }
+
+protected:
+  // Performs any action to undo side-effects of the execution if the callback
+  // has not yet been called (e.g. closing a file that was just opened).
+  // Not necessary for things that don't make persistent resources,
+  // e.g. cancelling a write does not have to undo the write.
+  virtual void onCancelledBeforeCallback(T){};
+
+  // Implementation of the actual action.
+  virtual T executeImpl() PURE;
+
+private:
+  std::function<void(T)> on_complete_;
+};
+
+} // namespace AsyncFiles
+} // namespace Common
+} // namespace Extensions
+} // namespace Envoy

source/extensions/common/async_files/async_file_context_base.cc

@@ -0,0 +1,27 @@
+#include "source/extensions/common/async_files/async_file_context_base.h"
+
+#include <functional>
+#include <memory>
+#include <utility>
+
+#include "source/extensions/common/async_files/async_file_action.h"
+#include "source/extensions/common/async_files/async_file_manager.h"
+
+#include "absl/base/thread_annotations.h"
+#include "absl/status/statusor.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Common {
+namespace AsyncFiles {
+
+AsyncFileContextBase::AsyncFileContextBase(AsyncFileManager& manager) : manager_(manager) {}
+
+std::function<void()> AsyncFileContextBase::enqueue(std::shared_ptr<AsyncFileAction> action) {
+  return manager_.enqueue(std::move(action));
+}
+
+} // namespace AsyncFiles
+} // namespace Common
+} // namespace Extensions
+} // namespace Envoy