python: keep stream and engine alive for callbacks

library/cc/BUILD

@@ -10,6 +10,7 @@ envoy_cc_library(
         "bridge_utility.cc",
         "engine.cc",
         "engine_builder.cc",
+        "engine_callbacks.cc",
         "headers.cc",
         "headers_builder.cc",
         "log_level.cc",
@@ -33,6 +34,7 @@ envoy_cc_library(
         "bridge_utility.h",
         "engine.h",
         "engine_builder.h",
+        "engine_callbacks.h",
         "envoy_error.h",
         "headers.h",
         "headers_builder.h",

library/cc/engine.cc

@@ -6,52 +6,26 @@
 namespace Envoy {
 namespace Platform {
 
-namespace {
-
-void c_on_engine_running(void* context) {
-  EngineCallbacks* engine_callbacks = static_cast<EngineCallbacks*>(context);
-  engine_callbacks->on_engine_running();
-}
-
-void c_on_exit(void* context) {
-  // NOTE: this function is intentionally empty
-  // as we don't actually do any post-processing on exit.
-  (void)context;
-}
-
-} // namespace
-
-Engine::Engine(envoy_engine_t engine, const std::string& configuration, LogLevel log_level,
-               EngineCallbacksSharedPtr callbacks)
-    : engine_(engine), callbacks_(callbacks), terminated_(false) {
-  envoy_engine_callbacks envoy_callbacks{
-      .on_engine_running = &c_on_engine_running,
-      .on_exit = &c_on_exit,
-      .context = this->callbacks_.get(),
-  };
-
-  envoy_logger null_logger{
-      .log = nullptr,
-      .release = envoy_noop_release,
-      .context = nullptr,
-  };
-
-  run_engine(this->engine_, envoy_callbacks, null_logger, configuration.c_str(),
-             log_level_to_string(log_level).c_str());
-
-  this->stream_client_ = std::make_shared<StreamClient>(this->engine_);
-  this->pulse_client_ = std::make_shared<PulseClient>();
+Engine::Engine(envoy_engine_t engine) : engine_(engine), terminated_(false) {}
+
+// we lazily construct the stream and pulse clients
+// because they either require or will require a weak ptr
+// which can't be provided from inside of the constructor
+// because of how std::enable_shared_from_this works
+StreamClientSharedPtr Engine::stream_client() {
+  if (!this->stream_client_) {
+    this->stream_client_ = std::make_shared<StreamClient>(this->weak_from_this());
+  }
+  return this->stream_client_;
 }
 
-Engine::~Engine() {
-  if (!this->terminated_) {
-    terminate_engine(this->engine_);
+PulseClientSharedPtr Engine::pulse_client() {
+  if (!this->pulse_client_) {
+    this->pulse_client_ = std::make_shared<PulseClient>();
   }
+  return this->pulse_client_;
 }
 
-StreamClientSharedPtr Engine::stream_client() { return this->stream_client_; }
-PulseClientSharedPtr Engine::pulse_client() { return this->pulse_client_; }
-
 void Engine::terminate() {
   if (this->terminated_) {
     throw std::runtime_error("attempting to double terminate Engine");

library/cc/engine.h

@@ -10,36 +10,25 @@
 namespace Envoy {
 namespace Platform {
 
-// TODO(crockeo): refactor engine callbacks
-//   - make EngineCallbacks struct with on_engine_running and (eventually) on_exit
-//   - change context from Engine ptr to EngineCallbacks ptr
-//   - move c_on_(...) from private static fn to static fn in anonymous namespace
-
-struct EngineCallbacks {
-  std::function<void()> on_engine_running;
-  // unused:
-  // std::function<void()> on_exit;
-};
-
-using EngineCallbacksSharedPtr = std::shared_ptr<EngineCallbacks>;
+class StreamClient;
+using StreamClientSharedPtr = std::shared_ptr<StreamClient>;
 
-class Engine {
+class Engine : public std::enable_shared_from_this<Engine> {
 public:
-  ~Engine();
-
   StreamClientSharedPtr stream_client();
   PulseClientSharedPtr pulse_client();
 
   void terminate();
 
 private:
-  Engine(envoy_engine_t engine, const std::string& configuration, LogLevel log_level,
-         EngineCallbacksSharedPtr callbacks);
+  Engine(envoy_engine_t engine);
 
+  // required to access private constructor
   friend class EngineBuilder;
+  // required to use envoy_engine_t without exposing it publicly
+  friend class StreamPrototype;
 
   envoy_engine_t engine_;
-  EngineCallbacksSharedPtr callbacks_;
   StreamClientSharedPtr stream_client_;
   PulseClientSharedPtr pulse_client_;
   bool terminated_;

library/cc/engine_builder.cc

@@ -10,7 +10,7 @@ EngineBuilder::EngineBuilder() : EngineBuilder(std::string(config_template)) {}
 
 EngineBuilder& EngineBuilder::add_log_level(LogLevel log_level) {
   this->log_level_ = log_level;
-  this->callbacks_ = std::make_shared<EngineCallbacks>();
+  this->callbacks_ = std::make_unique<EngineCallbacks>();
   return *this;
 }
 
@@ -89,8 +89,23 @@ EngineSharedPtr EngineBuilder::build() {
     }
   }
 
-  Engine* engine = new Engine(init_engine(), config_str, this->log_level_, this->callbacks_);
-  return EngineSharedPtr(engine);
+  auto envoy_engine = init_engine();
+  Engine* engine = new Engine(envoy_engine);
+  auto engine_ptr = EngineSharedPtr(engine);
+
+  auto callbacks = this->callbacks_.release();
+  callbacks->parent = engine_ptr;
+
+  envoy_logger null_logger{
+      .log = nullptr,
+      .release = envoy_noop_release,
+      .context = nullptr,
+  };
+
+  run_engine(envoy_engine, callbacks->as_envoy_engine_callbacks(), null_logger, config_str.c_str(),
+             log_level_to_string(this->log_level_).c_str());
+
+  return engine_ptr;
 }
 
 } // namespace Platform

library/cc/engine_builder.h

@@ -4,6 +4,7 @@
 #include <string>
 
 #include "engine.h"
+#include "engine_callbacks.h"
 #include "log_level.h"
 
 namespace Envoy {
@@ -36,7 +37,7 @@ class EngineBuilder {
 
 private:
   LogLevel log_level_ = LogLevel::info;
-  EngineCallbacksSharedPtr callbacks_;
+  EngineCallbacksPtr callbacks_;
 
   std::string config_template_;
   std::string stats_domain_ = "0.0.0.0";

library/cc/engine_callbacks.cc

@@ -0,0 +1,29 @@
+#include "engine_callbacks.h"
+
+namespace Envoy {
+namespace Platform {
+
+namespace {
+
+void c_on_engine_running(void* context) {
+  EngineCallbacks* engine_callbacks = static_cast<EngineCallbacks*>(context);
+  engine_callbacks->on_engine_running();
+}
+
+void c_on_exit(void* context) {
+  EngineCallbacks* engine_callbacks = static_cast<EngineCallbacks*>(context);
+  delete engine_callbacks;
+}
+
+} // namespace
+
+envoy_engine_callbacks EngineCallbacks::as_envoy_engine_callbacks() {
+  return envoy_engine_callbacks{
+      .on_engine_running = &c_on_engine_running,
+      .on_exit = &c_on_exit,
+      .context = this,
+  };
+}
+
+} // namespace Platform
+} // namespace Envoy

library/cc/engine_callbacks.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <functional>
+
+#include "engine.h"
+
+namespace Envoy {
+namespace Platform {
+
+struct EngineCallbacks {
+  std::function<void()> on_engine_running;
+  // unused:
+  // std::function<void()> on_exit;
+
+  envoy_engine_callbacks as_envoy_engine_callbacks();
+
+  EngineSharedPtr parent;
+};
+
+using EngineCallbacksPtr = std::unique_ptr<EngineCallbacks>;
+
+} // namespace Platform
+} // namespace Envoy

library/cc/stream.cc

@@ -1,16 +1,13 @@
 #include "stream.h"
 
-#include <iostream>
-
 #include "bridge_utility.h"
 #include "library/common/main_interface.h"
 #include "library/common/types/c_types.h"
 
 namespace Envoy {
 namespace Platform {
 
-Stream::Stream(envoy_stream_t handle, StreamCallbacksSharedPtr callbacks)
-    : handle_(handle), callbacks_(callbacks) {}
+Stream::Stream(EngineSharedPtr engine, envoy_stream_t handle) : engine_(engine), handle_(handle) {}
 
 Stream& Stream::send_headers(RequestHeadersSharedPtr headers, bool end_stream) {
   envoy_headers raw_headers = raw_header_map_as_envoy_headers(headers->all_headers());

library/cc/stream.h

@@ -10,9 +10,12 @@
 namespace Envoy {
 namespace Platform {
 
+class Engine;
+using EngineSharedPtr = std::shared_ptr<Engine>;
+
 class Stream {
 public:
-  Stream(envoy_stream_t handle, StreamCallbacksSharedPtr callbacks);
+  Stream(EngineSharedPtr engine_, envoy_stream_t handle);
 
   Stream& send_headers(RequestHeadersSharedPtr headers, bool end_stream);
   Stream& send_data(envoy_data data);
@@ -21,8 +24,8 @@ class Stream {
   void cancel();
 
 private:
+  EngineSharedPtr engine_;
   envoy_stream_t handle_;
-  StreamCallbacksSharedPtr callbacks_;
 };
 
 using StreamSharedPtr = std::shared_ptr<Stream>;

library/cc/stream_callbacks.cc

@@ -61,7 +61,8 @@ void* c_on_error(envoy_error raw_error, void* context) {
     auto on_error = stream_callbacks->on_error.value();
     on_error(error);
   }
-  return context;
+  delete stream_callbacks;
+  return nullptr;
 }
 
 void* c_on_complete(void* context) {
@@ -70,7 +71,8 @@ void* c_on_complete(void* context) {
     auto on_complete = stream_callbacks->on_complete.value();
     on_complete();
   }
-  return context;
+  delete stream_callbacks;
+  return nullptr;
 }
 
 void* c_on_cancel(void* context) {
@@ -79,7 +81,8 @@ void* c_on_cancel(void* context) {
     auto on_cancel = stream_callbacks->on_cancel.value();
     on_cancel();
   }
-  return context;
+  delete stream_callbacks;
+  return nullptr;
 }
 
 } // namespace

library/cc/stream_callbacks.h

@@ -9,10 +9,14 @@
 #include "library/common/types/c_types.h"
 #include "response_headers.h"
 #include "response_trailers.h"
+#include "stream.h"
 
 namespace Envoy {
 namespace Platform {
 
+class Stream;
+using StreamSharedPtr = std::shared_ptr<Stream>;
+
 using OnHeadersCallback = std::function<void(ResponseHeadersSharedPtr headers, bool end_stream)>;
 using OnDataCallback = std::function<void(envoy_data data, bool end_stream)>;
 using OnTrailersCallback = std::function<void(ResponseTrailersSharedPtr trailers)>;
@@ -29,6 +33,8 @@ struct StreamCallbacks {
   absl::optional<OnCancelCallback> on_cancel;
 
   envoy_http_callbacks as_envoy_http_callbacks();
+
+  StreamSharedPtr parent;
 };
 
 using StreamCallbacksSharedPtr = std::shared_ptr<StreamCallbacks>;

library/cc/stream_client.cc

@@ -3,10 +3,14 @@
 namespace Envoy {
 namespace Platform {
 
-StreamClient::StreamClient(envoy_engine_t engine) : engine_(engine) {}
+StreamClient::StreamClient(EngineWeakPtr engine) : engine_(engine) {}
 
 StreamPrototypeSharedPtr StreamClient::new_stream_prototype() {
-  return std::make_shared<StreamPrototype>(this->engine_);
+  if (auto engine = this->engine_.lock()) {
+    return std::make_shared<StreamPrototype>(engine);
+  } else {
+    throw std::runtime_error("attempted to use Engine weakptr after free");
+  }
 }
 
 } // namespace Platform

library/cc/stream_client.h

@@ -2,19 +2,26 @@
 
 #include <memory>
 
+#include "engine.h"
 #include "stream_prototype.h"
 
 namespace Envoy {
 namespace Platform {
 
+class Engine;
+using EngineWeakPtr = std::weak_ptr<Engine>;
+
+class StreamPrototype;
+using StreamPrototypeSharedPtr = std::shared_ptr<StreamPrototype>;
+
 class StreamClient {
 public:
-  StreamClient(envoy_engine_t engine);
+  StreamClient(EngineWeakPtr engine);
 
   StreamPrototypeSharedPtr new_stream_prototype();
 
 private:
-  envoy_engine_t engine_;
+  EngineWeakPtr engine_;
 };
 
 using StreamClientSharedPtr = std::shared_ptr<StreamClient>;