wasm: allow execution of multiple instances of the same plugin.

bazel/repository_locations.bzl

@@ -830,10 +830,10 @@ REPOSITORY_LOCATIONS_SPEC = dict(
         project_name = "WebAssembly for Proxies (C++ host implementation)",
         project_desc = "WebAssembly for Proxies (C++ host implementation)",
         project_url = "https://github.com/proxy-wasm/proxy-wasm-cpp-host",
-        version = "d54b3795e7c3e61015dac2c2110b0da2be999b8e",
-        sha256 = "e95ad57b6b550b039d4baa35c896f7f523e427b7278ba56f22fac6e1bef8c7f0",
+        version = "fa18f0005afd47e515577af7fdfe94b8a0268860",
+        sha256 = "8dfb533513fbc5bb2493bae1b8dc825bcc2c31776c6b1761b1713ac1f133df5a",
         strip_prefix = "proxy-wasm-cpp-host-{version}",
-        urls = ["https://github.com/proxy-wasm/proxy-wasm-cpp-host/archive/{version}.tar.gz"],
+        urls = ["https://github.com/PiotrSikora/proxy-wasm-cpp-host/archive/{version}.tar.gz"],
         use_category = ["dataplane_ext"],
         extensions = [
             "envoy.access_loggers.wasm",
@@ -842,7 +842,7 @@ REPOSITORY_LOCATIONS_SPEC = dict(
             "envoy.filters.network.wasm",
             "envoy.stat_sinks.wasm",
         ],
-        release_date = "2020-10-22",
+        release_date = "2020-10-29",
         cpe = "N/A",
     ),
     emscripten_toolchain = dict(

source/extensions/access_loggers/wasm/config.cc

@@ -23,8 +23,6 @@ WasmAccessLogFactory::createAccessLogInstance(const Protobuf::Message& proto_con
   const auto& config = MessageUtil::downcastAndValidate<
       const envoy::extensions::access_loggers::wasm::v3::WasmAccessLog&>(
       proto_config, context.messageValidationVisitor());
-  auto access_log =
-      std::make_shared<WasmAccessLog>(config.config().root_id(), nullptr, std::move(filter));
 
   // Create a base WASM to verify that the code loads before setting/cloning the for the
   // individual threads.
@@ -35,6 +33,8 @@ WasmAccessLogFactory::createAccessLogInstance(const Protobuf::Message& proto_con
       envoy::config::core::v3::TrafficDirection::UNSPECIFIED, context.localInfo(),
       nullptr /* listener_metadata */);
 
+  auto access_log = std::make_shared<WasmAccessLog>(plugin, nullptr, std::move(filter));
+
   auto callback = [access_log, &context, plugin](Common::Wasm::WasmHandleSharedPtr base_wasm) {
     auto tls_slot = context.threadLocal().allocateSlot();
 

source/extensions/access_loggers/wasm/wasm_access_log_impl.h

@@ -12,13 +12,15 @@ namespace Extensions {
 namespace AccessLoggers {
 namespace Wasm {
 
+using Envoy::Extensions::Common::Wasm::PluginSharedPtr;
 using Envoy::Extensions::Common::Wasm::WasmHandle;
 
 class WasmAccessLog : public AccessLog::Instance {
 public:
-  WasmAccessLog(absl::string_view root_id, ThreadLocal::SlotPtr tls_slot,
+  WasmAccessLog(const PluginSharedPtr& plugin, ThreadLocal::SlotPtr tls_slot,
                 AccessLog::FilterPtr filter)
-      : root_id_(root_id), tls_slot_(std::move(tls_slot)), filter_(std::move(filter)) {}
+      : plugin_(plugin), tls_slot_(std::move(tls_slot)), filter_(std::move(filter)) {}
+
   void log(const Http::RequestHeaderMap* request_headers,
            const Http::ResponseHeaderMap* response_headers,
            const Http::ResponseTrailerMap* response_trailers,
@@ -31,7 +33,7 @@ class WasmAccessLog : public AccessLog::Instance {
     }
 
     if (tls_slot_->get()) {
-      tls_slot_->getTyped<WasmHandle>().wasm()->log(root_id_, request_headers, response_headers,
+      tls_slot_->getTyped<WasmHandle>().wasm()->log(plugin_, request_headers, response_headers,
                                                     response_trailers, stream_info);
     }
   }
@@ -41,8 +43,18 @@ class WasmAccessLog : public AccessLog::Instance {
     tls_slot_ = std::move(tls_slot);
   }
 
+  ~WasmAccessLog() {
+    if (tls_slot_->get()) {
+      tls_slot_->runOnAllThreads([plugin = plugin_](ThreadLocal::ThreadLocalObjectSharedPtr object)
+                                     -> ThreadLocal::ThreadLocalObjectSharedPtr {
+        object->asType<WasmHandle>().wasm()->startShutdown(plugin);
+        return object;
+      });
+    }
+  }
+
 private:
-  std::string root_id_;
+  Common::Wasm::PluginSharedPtr plugin_;
   ThreadLocal::SlotPtr tls_slot_;
   AccessLog::FilterPtr filter_;
 };

source/extensions/bootstrap/wasm/config.cc

@@ -48,7 +48,7 @@ void WasmFactory::createWasm(const envoy::extensions::wasm::v3::WasmService& con
       return std::static_pointer_cast<ThreadLocal::ThreadLocalObject>(
           Common::Wasm::getOrCreateThreadLocalWasm(base_wasm, plugin, dispatcher));
     });
-    cb(std::make_unique<WasmService>(std::move(tls_slot)));
+    cb(std::make_unique<WasmService>(plugin, std::move(tls_slot)));
   };
 
   if (!Common::Wasm::createWasm(

source/extensions/bootstrap/wasm/config.h

@@ -19,9 +19,21 @@ namespace Wasm {
 class WasmService {
 public:
   WasmService(Common::Wasm::WasmHandleSharedPtr singleton) : singleton_(std::move(singleton)) {}
-  WasmService(ThreadLocal::SlotPtr tls_slot) : tls_slot_(std::move(tls_slot)) {}
+  WasmService(Common::Wasm::PluginSharedPtr plugin, ThreadLocal::SlotPtr tls_slot)
+      : plugin_(plugin), tls_slot_(std::move(tls_slot)) {}
+
+  ~WasmService() {
+    if (tls_slot_ && tls_slot_->get()) {
+      tls_slot_->runOnAllThreads([plugin = plugin_](ThreadLocal::ThreadLocalObjectSharedPtr object)
+                                     -> ThreadLocal::ThreadLocalObjectSharedPtr {
+        object->asType<Common::Wasm::WasmHandle>().wasm()->startShutdown(plugin);
+        return object;
+      });
+    }
+  }
 
 private:
+  Common::Wasm::PluginSharedPtr plugin_;
   Common::Wasm::WasmHandleSharedPtr singleton_;
   ThreadLocal::SlotPtr tls_slot_;
 };

source/extensions/common/wasm/context.cc

@@ -810,8 +810,8 @@ BufferInterface* Context::getBuffer(WasmBufferType type) {
   case WasmBufferType::VmConfiguration:
     return buffer_.set(wasm()->vm_configuration());
   case WasmBufferType::PluginConfiguration:
-    if (plugin_) {
-      return buffer_.set(plugin_->plugin_configuration_);
+    if (temp_plugin_) {
+      return buffer_.set(temp_plugin_->plugin_configuration_);
     }
     return nullptr;
   case WasmBufferType::HttpRequestBody:
@@ -1182,18 +1182,18 @@ bool Context::validateConfiguration(absl::string_view configuration,
   if (!wasm()->validate_configuration_) {
     return true;
   }
-  plugin_ = plugin_base;
+  temp_plugin_ = plugin_base;
   auto result =
       wasm()
           ->validate_configuration_(this, id_, static_cast<uint32_t>(configuration.size()))
           .u64_ != 0;
-  plugin_.reset();
+  temp_plugin_.reset();
   return result;
 }
 
 absl::string_view Context::getConfiguration() {
-  if (plugin_) {
-    return plugin_->plugin_configuration_;
+  if (temp_plugin_) {
+    return temp_plugin_->plugin_configuration_;
   } else {
     return wasm()->vm_configuration();
   }

source/extensions/common/wasm/wasm.cc

@@ -243,16 +243,16 @@ ContextBase* Wasm::createRootContext(const std::shared_ptr<PluginBase>& plugin)
 
 ContextBase* Wasm::createVmContext() { return new Context(this); }
 
-void Wasm::log(absl::string_view root_id, const Http::RequestHeaderMap* request_headers,
+void Wasm::log(const PluginSharedPtr& plugin, const Http::RequestHeaderMap* request_headers,
                const Http::ResponseHeaderMap* response_headers,
                const Http::ResponseTrailerMap* response_trailers,
                const StreamInfo::StreamInfo& stream_info) {
-  auto context = getRootContext(root_id);
+  auto context = getRootContext(plugin, true);
   context->log(request_headers, response_headers, response_trailers, stream_info);
 }
 
-void Wasm::onStatsUpdate(absl::string_view root_id, Envoy::Stats::MetricSnapshot& snapshot) {
-  auto context = getRootContext(root_id);
+void Wasm::onStatsUpdate(const PluginSharedPtr& plugin, Envoy::Stats::MetricSnapshot& snapshot) {
+  auto context = getRootContext(plugin, true);
   context->onStatsUpdate(snapshot);
 }
 

source/extensions/common/wasm/wasm.h

@@ -54,8 +54,8 @@ class Wasm : public WasmBase, Logger::Loggable<Logger::Id::wasm> {
 
   Upstream::ClusterManager& clusterManager() const { return cluster_manager_; }
   Event::Dispatcher& dispatcher() { return dispatcher_; }
-  Context* getRootContext(absl::string_view root_id) {
-    return static_cast<Context*>(WasmBase::getRootContext(root_id));
+  Context* getRootContext(const std::shared_ptr<PluginBase>& plugin, bool allow_closed) {
+    return static_cast<Context*>(WasmBase::getRootContext(plugin, allow_closed));
   }
   void setTimerPeriod(uint32_t root_context_id, std::chrono::milliseconds period) override;
   virtual void tickHandler(uint32_t root_context_id);
@@ -72,12 +72,13 @@ class Wasm : public WasmBase, Logger::Loggable<Logger::Id::wasm> {
   void getFunctions() override;
 
   // AccessLog::Instance
-  void log(absl::string_view root_id, const Http::RequestHeaderMap* request_headers,
+  void log(const PluginSharedPtr& plugin, const Http::RequestHeaderMap* request_headers,
            const Http::ResponseHeaderMap* response_headers,
            const Http::ResponseTrailerMap* response_trailers,
            const StreamInfo::StreamInfo& stream_info);
 
-  void onStatsUpdate(absl::string_view root_id, Envoy::Stats::MetricSnapshot& snapshot);
+  void onStatsUpdate(const PluginSharedPtr& plugin, Envoy::Stats::MetricSnapshot& snapshot);
+
   virtual std::string buildVersion() { return BUILD_VERSION_NUMBER; }
 
   void initializeLifecycle(Server::ServerLifecycleNotifier& lifecycle_notifier);

source/extensions/filters/http/wasm/wasm_filter.cc

@@ -1,14 +1,5 @@
 #include "extensions/filters/http/wasm/wasm_filter.h"
 
-#include "envoy/http/codes.h"
-
-#include "common/buffer/buffer_impl.h"
-#include "common/common/assert.h"
-#include "common/common/enum_to_int.h"
-#include "common/http/header_map_impl.h"
-#include "common/http/message_impl.h"
-#include "common/http/utility.h"
-
 namespace Envoy {
 namespace Extensions {
 namespace HttpFilters {
@@ -46,6 +37,16 @@ FilterConfig::FilterConfig(const envoy::extensions::filters::http::wasm::v3::Was
   }
 }
 
+FilterConfig::~FilterConfig() {
+  if (tls_slot_->get()) {
+    tls_slot_->runOnAllThreads([plugin = plugin_](ThreadLocal::ThreadLocalObjectSharedPtr object)
+                                   -> ThreadLocal::ThreadLocalObjectSharedPtr {
+      object->asType<WasmHandle>().wasm()->startShutdown(plugin);
+      return object;
+    });
+  }
+}
+
 } // namespace Wasm
 } // namespace HttpFilters
 } // namespace Extensions

source/extensions/filters/http/wasm/wasm_filter.h

@@ -23,6 +23,7 @@ class FilterConfig : Logger::Loggable<Logger::Id::wasm> {
 public:
   FilterConfig(const envoy::extensions::filters::http::wasm::v3::Wasm& proto_config,
                Server::Configuration::FactoryContext& context);
+  ~FilterConfig();
 
   std::shared_ptr<Context> createFilter() {
     Wasm* wasm = nullptr;
@@ -33,7 +34,7 @@ class FilterConfig : Logger::Loggable<Logger::Id::wasm> {
       return nullptr;
     }
     if (wasm && !root_context_id_) {
-      root_context_id_ = wasm->getRootContext(plugin_->root_id_)->id();
+      root_context_id_ = wasm->getRootContext(plugin_, false)->id();
     }
     return std::make_shared<Context>(wasm, root_context_id_, plugin_);
   }

source/extensions/filters/network/wasm/wasm_filter.cc

@@ -1,9 +1,5 @@
 #include "extensions/filters/network/wasm/wasm_filter.h"
 
-#include "common/buffer/buffer_impl.h"
-#include "common/common/assert.h"
-#include "common/common/enum_to_int.h"
-
 namespace Envoy {
 namespace Extensions {
 namespace NetworkFilters {
@@ -41,6 +37,16 @@ FilterConfig::FilterConfig(const envoy::extensions::filters::network::wasm::v3::
   }
 }
 
+FilterConfig::~FilterConfig() {
+  if (tls_slot_->get()) {
+    tls_slot_->runOnAllThreads([plugin = plugin_](ThreadLocal::ThreadLocalObjectSharedPtr object)
+                                   -> ThreadLocal::ThreadLocalObjectSharedPtr {
+      object->asType<WasmHandle>().wasm()->startShutdown(plugin);
+      return object;
+    });
+  }
+}
+
 } // namespace Wasm
 } // namespace NetworkFilters
 } // namespace Extensions

source/extensions/filters/network/wasm/wasm_filter.h

@@ -23,6 +23,7 @@ class FilterConfig : Logger::Loggable<Logger::Id::wasm> {
 public:
   FilterConfig(const envoy::extensions::filters::network::wasm::v3::Wasm& proto_config,
                Server::Configuration::FactoryContext& context);
+  ~FilterConfig();
 
   std::shared_ptr<Context> createFilter() {
     Wasm* wasm = nullptr;
@@ -33,11 +34,12 @@ class FilterConfig : Logger::Loggable<Logger::Id::wasm> {
       return nullptr;
     }
     if (wasm && !root_context_id_) {
-      root_context_id_ = wasm->getRootContext(plugin_->root_id_)->id();
+      root_context_id_ = wasm->getRootContext(plugin_, false)->id();
     }
     return std::make_shared<Context>(wasm, root_context_id_, plugin_);
   }
-  Envoy::Extensions::Common::Wasm::Wasm* wasm() {
+
+  Envoy::Extensions::Common::Wasm::Wasm* wasmForTest() {
     return tls_slot_->getTyped<WasmHandle>().wasm().get();
   }
 

source/extensions/stat_sinks/wasm/config.cc

@@ -22,14 +22,14 @@ WasmSinkFactory::createStatsSink(const Protobuf::Message& proto_config,
       MessageUtil::downcastAndValidate<const envoy::extensions::stat_sinks::wasm::v3::Wasm&>(
           proto_config, context.messageValidationContext().staticValidationVisitor());
 
-  auto wasm_sink = std::make_unique<WasmStatSink>(config.config().root_id(), nullptr);
-
   auto plugin = std::make_shared<Common::Wasm::Plugin>(
       config.config().name(), config.config().root_id(), config.config().vm_config().vm_id(),
       config.config().vm_config().runtime(),
       Common::Wasm::anyToBytes(config.config().configuration()), config.config().fail_open(),
       envoy::config::core::v3::TrafficDirection::UNSPECIFIED, context.localInfo(), nullptr);
 
+  auto wasm_sink = std::make_unique<WasmStatSink>(plugin, nullptr);
+
   auto callback = [&wasm_sink, &context, plugin](Common::Wasm::WasmHandleSharedPtr base_wasm) {
     if (!base_wasm) {
       if (plugin->fail_open_) {

source/extensions/stat_sinks/wasm/wasm_stat_sink_impl.h

@@ -9,15 +9,16 @@ namespace Extensions {
 namespace StatSinks {
 namespace Wasm {
 
+using Envoy::Extensions::Common::Wasm::PluginSharedPtr;
 using Envoy::Extensions::Common::Wasm::WasmHandle;
 
 class WasmStatSink : public Stats::Sink {
 public:
-  WasmStatSink(absl::string_view root_id, Common::Wasm::WasmHandleSharedPtr singleton)
-      : root_id_(root_id), singleton_(std::move(singleton)) {}
+  WasmStatSink(const PluginSharedPtr& plugin, Common::Wasm::WasmHandleSharedPtr singleton)
+      : plugin_(plugin), singleton_(std::move(singleton)) {}
 
   void flush(Stats::MetricSnapshot& snapshot) override {
-    singleton_->wasm()->onStatsUpdate(root_id_, snapshot);
+    singleton_->wasm()->onStatsUpdate(plugin_, snapshot);
   }
 
   void setSingleton(Common::Wasm::WasmHandleSharedPtr singleton) {
@@ -31,7 +32,7 @@ class WasmStatSink : public Stats::Sink {
   }
 
 private:
-  std::string root_id_;
+  Common::Wasm::PluginSharedPtr plugin_;
   Common::Wasm::WasmHandleSharedPtr singleton_;
 };
 

test/extensions/bootstrap/wasm/wasm_test.cc

@@ -207,7 +207,7 @@ TEST_P(WasmTest, Segv) {
   auto context = static_cast<TestContext*>(wasm_->start(plugin_));
   EXPECT_CALL(*context, log_(spdlog::level::err, Eq("before badptr")));
   EXPECT_FALSE(wasm_->configure(context, plugin_));
-  wasm_->isFailed();
+  EXPECT_TRUE(wasm_->isFailed());
 }
 
 TEST_P(WasmTest, DivByZero) {
@@ -219,7 +219,7 @@ TEST_P(WasmTest, DivByZero) {
   auto context = static_cast<TestContext*>(wasm_->start(plugin_));
   EXPECT_CALL(*context, log_(spdlog::level::err, Eq("before div by zero")));
   context->onLog();
-  wasm_->isFailed();
+  EXPECT_TRUE(wasm_->isFailed());
 }
 
 TEST_P(WasmTest, IntrinsicGlobals) {

test/extensions/common/wasm/test_data/test_cpp.cc

@@ -267,6 +267,10 @@ WASM_EXPORT(uint32_t, proxy_on_done, (uint32_t)) {
   return 0;
 }
 
+WASM_EXPORT(void, proxy_on_tick, (uint32_t)) {
+  proxy_done();
+}
+
 WASM_EXPORT(void, proxy_on_delete, (uint32_t)) {
   std::string message = "on_delete logging";
   proxy_log(LogLevel::info, message.c_str(), message.size());