[utility]: wrap try in an assertion to ensure that exception are only caught on main thread

source/common/access_log/access_log_manager_impl.cc

@@ -2,6 +2,8 @@
 
 #include <string>
 
+#include "envoy/common/exception.h"
+
 #include "common/common/assert.h"
 #include "common/common/fmt.h"
 #include "common/common/lock_guard.h"
@@ -50,7 +52,11 @@ AccessLogFileImpl::AccessLogFileImpl(Filesystem::FilePtr&& file, Event::Dispatch
       })),
       thread_factory_(thread_factory), flush_interval_msec_(flush_interval_msec), stats_(stats) {
   flush_timer_->enableTimer(flush_interval_msec_);
-  open();
+  auto open_result = open();
+  if (!open_result.rc_) {
+    throw EnvoyException(fmt::format("unable to open file '{}': {}", file_->path(),
+                                     open_result.err_->getErrorDetails()));
+  }
 }
 
 Filesystem::FlagSet AccessLogFileImpl::defaultFlags() {
@@ -61,12 +67,9 @@ Filesystem::FlagSet AccessLogFileImpl::defaultFlags() {
   return default_flags;
 }
 
-void AccessLogFileImpl::open() {
-  const Api::IoCallBoolResult result = file_->open(defaultFlags());
-  if (!result.rc_) {
-    throw EnvoyException(
-        fmt::format("unable to open file '{}': {}", file_->path(), result.err_->getErrorDetails()));
-  }
+Api::IoCallBoolResult AccessLogFileImpl::open() {
+  Api::IoCallBoolResult result = file_->open(defaultFlags());
+  return result;
 }
 
 void AccessLogFileImpl::reopen() { reopen_file_ = true; }
@@ -87,7 +90,6 @@ AccessLogFileImpl::~AccessLogFileImpl() {
     if (flush_buffer_.length() > 0) {
       doWrite(flush_buffer_);
     }
-
     const Api::IoCallBoolResult result = file_->close();
     ASSERT(result.rc_, fmt::format("unable to close file '{}': {}", file_->path(),
                                    result.err_->getErrorDetails()));
@@ -149,19 +151,18 @@ void AccessLogFileImpl::flushThreadFunc() {
 
     // if we failed to open file before, then simply ignore
     if (file_->isOpen()) {
-      try {
-        if (reopen_file_) {
-          reopen_file_ = false;
-          const Api::IoCallBoolResult result = file_->close();
-          ASSERT(result.rc_, fmt::format("unable to close file '{}': {}", file_->path(),
-                                         result.err_->getErrorDetails()));
-          open();
+      if (reopen_file_) {
+        reopen_file_ = false;
+        const Api::IoCallBoolResult result = file_->close();
+        ASSERT(result.rc_, fmt::format("unable to close file '{}': {}", file_->path(),
+                                       result.err_->getErrorDetails()));
+        const Api::IoCallBoolResult open_result = open();
+        if (!open_result.rc_) {
+          stats_.reopen_failed_.inc();
+          return;
         }
-
-        doWrite(about_to_write_buffer_);
-      } catch (const EnvoyException&) {
-        stats_.reopen_failed_.inc();
       }
+      doWrite(about_to_write_buffer_);
     }
   }
 }

source/common/access_log/access_log_manager_impl.h

@@ -84,7 +84,7 @@ class AccessLogFileImpl : public AccessLogFile {
 private:
   void doWrite(Buffer::Instance& buffer);
   void flushThreadFunc();
-  void open();
+  Api::IoCallBoolResult open();
   void createFlushStructures();
 
   // return default flags set which used by open

source/common/common/BUILD

@@ -326,6 +326,7 @@ envoy_cc_library(
 
 envoy_cc_library(
     name = "thread_lib",
+    srcs = ["thread.cc"],
     hdrs = ["thread.h"],
     external_deps = ["abseil_synchronization"],
     deps = envoy_cc_platform_dep("thread_impl_lib") + [

source/common/common/thread.cc

@@ -0,0 +1,36 @@
+#include "common/common/thread.h"
+
+namespace Envoy {
+namespace Thread {
+
+bool MainThread::isMainThread() {
+  // If threading is off, only main thread is running.
+  auto main_thread_singleton = MainThreadSingleton::getExisting();
+  if (main_thread_singleton == nullptr) {
+    return true;
+  }
+  // When threading is on, compare thread id with main thread id.
+  return main_thread_singleton->inMainThread() || main_thread_singleton->inTestThread();
+}
+
+void MainThread::clear() {
+  delete MainThreadSingleton::getExisting();
+  MainThreadSingleton::clear();
+}
+
+void MainThread::initTestThread() {
+  if (!initialized()) {
+    MainThreadSingleton::initialize(new MainThread());
+  }
+  MainThreadSingleton::get().registerTestThread();
+}
+
+void MainThread::initMainThread() {
+  if (!initialized()) {
+    MainThreadSingleton::initialize(new MainThread());
+  }
+  MainThreadSingleton::get().registerMainThread();
+}
+
+} // namespace Thread
+} // namespace Envoy

source/common/common/thread.h

@@ -172,23 +172,46 @@ class AtomicPtr : private AtomicPtrArray<T, 1, alloc_mode> {
 struct MainThread {
   using MainThreadSingleton = InjectableSingleton<MainThread>;
   bool inMainThread() const { return main_thread_id_ == std::this_thread::get_id(); }
-  static void init() { MainThreadSingleton::initialize(new MainThread()); }
-  static void clear() {
-    delete MainThreadSingleton::getExisting();
-    MainThreadSingleton::clear();
-  }
-  static bool isMainThread() {
-    // If threading is off, only main thread is running.
-    if (MainThreadSingleton::getExisting() == nullptr) {
-      return true;
-    }
-    // When threading is on, compare thread id with main thread id.
-    return MainThreadSingleton::get().inMainThread();
+  bool inTestThread() const {
+    return test_thread_id_.has_value() && (test_thread_id_.value() == std::this_thread::get_id());
   }
+  void registerTestThread() { test_thread_id_ = std::this_thread::get_id(); }
+  void registerMainThread() { main_thread_id_ = std::this_thread::get_id(); }
+  static bool initialized() { return MainThreadSingleton::getExisting() != nullptr; }
+  /*
+   * Register the main thread id, should be called in main thread before threading is on. Currently
+   * called in ThreadLocal::InstanceImpl().
+   */
+  static void initMainThread();
+  /*
+   * Register the test thread id, should be called in test thread before threading is on. Allow
+   * some main thread only code to be executed on test thread.
+   */
+  static void initTestThread();
+  /*
+   * Delete the main thread singleton, should be called in main thread after threading
+   * has been shut down. Currently called in ~ThreadLocal::InstanceImpl().
+   */
+  static void clear();
+  static bool isMainThread();
 
 private:
-  std::thread::id main_thread_id_{std::this_thread::get_id()};
+  std::thread::id main_thread_id_;
+  absl::optional<std::thread::id> test_thread_id_;
 };
 
+// To improve exception safety in data plane, we plan to forbid the use of raw try in the core code
+// base. This macros uses main thread assertion to make sure that exceptions aren't thrown from
+// worker thread.
+#define TRY_ASSERT_MAIN_THREAD                                                                     \
+  try {                                                                                            \
+    ASSERT(Thread::MainThread::isMainThread());
+
+#define END_TRY }
+
+// TODO(chaoqinli-1123): Remove this macros after we have removed all the exceptions from data
+// plane.
+#define TRY_NEEDS_AUDIT try
+
 } // namespace Thread
 } // namespace Envoy

source/common/config/delta_subscription_state.cc

@@ -67,9 +67,9 @@ UpdateAck DeltaSubscriptionState::handleResponse(
   // We *always* copy the response's nonce into the next request, even if we're going to make that
   // request a NACK by setting error_detail.
   UpdateAck ack(message.nonce(), type_url_);
-  try {
-    handleGoodResponse(message);
-  } catch (const EnvoyException& e) {
+  TRY_ASSERT_MAIN_THREAD { handleGoodResponse(message); }
+  END_TRY
+  catch (const EnvoyException& e) {
     handleBadResponse(e, ack);
   }
   return ack;

source/common/config/filesystem_subscription_impl.cc

@@ -61,17 +61,20 @@ void FilesystemSubscriptionImpl::refresh() {
   ENVOY_LOG(debug, "Filesystem config refresh for {}", path_);
   stats_.update_attempt_.inc();
   ProtobufTypes::MessagePtr config_update;
-  try {
+  TRY_ASSERT_MAIN_THREAD {
     const std::string version = refreshInternal(&config_update);
     stats_.update_time_.set(DateUtil::nowToMilliseconds(api_.timeSource()));
     stats_.version_.set(HashUtil::xxHash64(version));
     stats_.version_text_.set(version);
     stats_.update_success_.inc();
     ENVOY_LOG(debug, "Filesystem config update accepted for {}: {}", path_,
               config_update->DebugString());
-  } catch (const ProtobufMessage::UnknownProtoFieldException& e) {
+  }
+  END_TRY
+  catch (const ProtobufMessage::UnknownProtoFieldException& e) {
     configRejected(e, config_update == nullptr ? "" : config_update->DebugString());
-  } catch (const EnvoyException& e) {
+  }
+  catch (const EnvoyException& e) {
     if (config_update != nullptr) {
       configRejected(e, config_update->DebugString());
     } else {

source/common/config/grpc_mux_impl.cc

@@ -198,7 +198,7 @@ void GrpcMuxImpl::onDiscoveryResponse(
   // the delta state. The proper fix for this is to converge these implementations,
   // see https://github.com/envoyproxy/envoy/issues/11477.
   same_type_resume = pause(type_url);
-  try {
+  TRY_ASSERT_MAIN_THREAD {
     // To avoid O(n^2) explosion (e.g. when we have 1000s of EDS watches), we
     // build a map here from resource name to resource and then walk watches_.
     // We have to walk all watches (and need an efficient map as a result) to
@@ -263,7 +263,9 @@ void GrpcMuxImpl::onDiscoveryResponse(
     // would do that tracking here.
     apiStateFor(type_url).request_.set_version_info(message->version_info());
     Memory::Utils::tryShrinkHeap();
-  } catch (const EnvoyException& e) {
+  }
+  END_TRY
+  catch (const EnvoyException& e) {
     for (auto watch : apiStateFor(type_url).watches_) {
       watch->callbacks_.onConfigUpdateFailed(
           Envoy::Config::ConfigUpdateFailureReason::UpdateRejected, &e);

source/common/config/http_subscription_impl.cc

@@ -82,13 +82,15 @@ void HttpSubscriptionImpl::createRequest(Http::RequestMessage& request) {
 void HttpSubscriptionImpl::parseResponse(const Http::ResponseMessage& response) {
   disableInitFetchTimeoutTimer();
   envoy::service::discovery::v3::DiscoveryResponse message;
-  try {
+  TRY_ASSERT_MAIN_THREAD {
     MessageUtil::loadFromJson(response.bodyAsString(), message, validation_visitor_);
-  } catch (const EnvoyException& e) {
+  }
+  END_TRY
+  catch (const EnvoyException& e) {
     handleFailure(Config::ConfigUpdateFailureReason::UpdateRejected, &e);
     return;
   }
-  try {
+  TRY_ASSERT_MAIN_THREAD {
     const auto decoded_resources =
         DecodedResourcesWrapper(resource_decoder_, message.resources(), message.version_info());
     callbacks_.onConfigUpdate(decoded_resources.refvec_, message.version_info());
@@ -97,7 +99,9 @@ void HttpSubscriptionImpl::parseResponse(const Http::ResponseMessage& response)
     stats_.version_.set(HashUtil::xxHash64(request_.version_info()));
     stats_.version_text_.set(request_.version_info());
     stats_.update_success_.inc();
-  } catch (const EnvoyException& e) {
+  }
+  END_TRY
+  catch (const EnvoyException& e) {
     handleFailure(Config::ConfigUpdateFailureReason::UpdateRejected, &e);
   }
 }

source/common/filter/http/filter_config_discovery_impl.cc

@@ -3,6 +3,7 @@
 #include "envoy/config/core/v3/extension.pb.validate.h"
 #include "envoy/server/filter_config.h"
 
+#include "common/common/thread.h"
 #include "common/config/utility.h"
 #include "common/grpc/common.h"
 #include "common/protobuf/utility.h"
@@ -230,10 +231,11 @@ FilterConfigProviderPtr FilterConfigProviderManagerImpl::createDynamicFilterConf
   // and the applied config eventually converges once ECDS update arrives.
   bool last_config_valid = false;
   if (subscription->lastConfig().has_value()) {
-    try {
+    TRY_ASSERT_MAIN_THREAD {
       provider->validateTypeUrl(subscription->lastTypeUrl());
       last_config_valid = true;
-    } catch (const EnvoyException& e) {
+    }
+    END_TRY catch (const EnvoyException& e) {
       ENVOY_LOG(debug, "ECDS subscription {} is invalid in a listener context: {}.",
                 filter_config_name, e.what());
       subscription->incrementConflictCounter();

source/common/formatter/substitution_formatter.cc

@@ -13,6 +13,7 @@
 #include "common/common/assert.h"
 #include "common/common/empty_string.h"
 #include "common/common/fmt.h"
+#include "common/common/thread.h"
 #include "common/common/utility.h"
 #include "common/config/metadata.h"
 #include "common/grpc/common.h"
@@ -1294,9 +1295,9 @@ ProtobufWkt::Value FilterStateFormatter::formatValue(const Http::RequestHeaderMa
   }
 
   ProtobufWkt::Value val;
-  try {
-    MessageUtil::jsonConvertValue(*proto, val);
-  } catch (EnvoyException& ex) {
+  // TODO(chaoqin-li1123): make this conversion return an error status instead of throwing.
+  TRY_NEEDS_AUDIT { MessageUtil::jsonConvertValue(*proto, val); }
+  catch (EnvoyException& ex) {
     return unspecifiedValue();
   }
   return val;

source/common/http/rest_api_fetcher.cc

@@ -39,9 +39,9 @@ void RestApiFetcher::onSuccess(const Http::AsyncClient::Request& request,
     return;
   }
 
-  try {
-    parseResponse(*response);
-  } catch (EnvoyException& e) {
+  TRY_ASSERT_MAIN_THREAD { parseResponse(*response); }
+  END_TRY
+  catch (EnvoyException& e) {
     onFetchFailure(Config::ConfigUpdateFailureReason::UpdateRejected, &e);
   }
 

source/common/http/utility.cc

@@ -39,17 +39,20 @@ namespace Utility {
 Http::Status exceptionToStatus(std::function<Http::Status(Buffer::Instance&)> dispatch,
                                Buffer::Instance& data) {
   Http::Status status;
-  try {
+  TRY_NEEDS_AUDIT {
     status = dispatch(data);
     // TODO(#10878): Remove this when exception removal is complete. It is currently in migration,
     // so dispatch may either return an error status or throw an exception. Soon we won't need to
     // catch these exceptions, as all codec errors will be migrated to using error statuses that are
     // returned from dispatch.
-  } catch (FrameFloodException& e) {
+  }
+  catch (FrameFloodException& e) {
     status = bufferFloodError(e.what());
-  } catch (CodecProtocolException& e) {
+  }
+  catch (CodecProtocolException& e) {
     status = codecProtocolError(e.what());
-  } catch (PrematureResponseException& e) {
+  }
+  catch (PrematureResponseException& e) {
     status = prematureResponseError(e.what(), e.responseCode());
   }
   return status;
@@ -627,17 +630,16 @@ Utility::getLastAddressFromXFF(const Http::RequestHeaderMap& request_headers,
   xff_string = StringUtil::ltrim(xff_string);
   xff_string = StringUtil::rtrim(xff_string);
 
-  try {
-    // This technically requires a copy because inet_pton takes a null terminated string. In
-    // practice, we are working with a view at the end of the owning string, and could pass the
-    // raw pointer.
-    // TODO(mattklein123) PERF: Avoid the copy here.
-    return {
-        Network::Utility::parseInternetAddress(std::string(xff_string.data(), xff_string.size())),
-        last_comma == std::string::npos && num_to_skip == 0};
-  } catch (const EnvoyException&) {
-    return {nullptr, false};
+  // This technically requires a copy because inet_pton takes a null terminated string. In
+  // practice, we are working with a view at the end of the owning string, and could pass the
+  // raw pointer.
+  // TODO(mattklein123) PERF: Avoid the copy here.
+  Network::Address::InstanceConstSharedPtr address = Network::Utility::parseInternetAddressNoThrow(
+      std::string(xff_string.data(), xff_string.size()));
+  if (address != nullptr) {
+    return {address, last_comma == std::string::npos && num_to_skip == 0};
   }
+  return {nullptr, false};
 }
 
 bool Utility::sanitizeConnectionHeader(Http::RequestHeaderMap& headers) {