[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

@@ -149,7 +149,7 @@ void AccessLogFileImpl::flushThreadFunc() {
 
     // if we failed to open file before, then simply ignore
     if (file_->isOpen()) {
-      try {
+      TRY_NEEDS_AUDIT {
         if (reopen_file_) {
           reopen_file_ = false;
           const Api::IoCallBoolResult result = file_->close();
@@ -159,7 +159,8 @@ void AccessLogFileImpl::flushThreadFunc() {
         }
 
         doWrite(about_to_write_buffer_);
-      } catch (const EnvoyException&) {
+      }
+      catch (const EnvoyException&) {
         stats_.reopen_failed_.inc();
       }
     }

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,35 @@
+#include "common/common/thread.h"
+
+namespace Envoy {
+namespace Thread {
+
+bool MainThread::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() || MainThreadSingleton::get().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,34 @@ 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; }
+  static void initMainThread();
+  static void initTestThread();
+  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,8 @@ ProtobufWkt::Value FilterStateFormatter::formatValue(const Http::RequestHeaderMa
   }
 
   ProtobufWkt::Value val;
-  try {
-    MessageUtil::jsonConvertValue(*proto, val);
-  } catch (EnvoyException& ex) {
+  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) {

source/common/network/apple_dns_impl.cc

@@ -148,15 +148,17 @@ ActiveDnsQuery* AppleDnsResolverImpl::resolve(const std::string& dns_name,
   ENVOY_LOG(debug, "DNS resolver resolve={}", dns_name);
 
   Address::InstanceConstSharedPtr address{};
-  try {
+  TRY_ASSERT_MAIN_THREAD {
     // When an IP address is submitted to c-ares in DnsResolverImpl, c-ares synchronously returns
     // the IP without submitting a DNS query. Because Envoy has come to rely on this behavior, this
     // resolver implements a similar resolution path to avoid making improper DNS queries for
     // resolved IPs.
     address = Utility::parseInternetAddress(dns_name);
     ENVOY_LOG(debug, "DNS resolver resolved ({}) to ({}) without issuing call to Apple API",
               dns_name, address->asString());
-  } catch (const EnvoyException& e) {
+  }
+  END_TRY
+  catch (const EnvoyException& e) {
     // Resolution via Apple APIs
     ENVOY_LOG(trace, "DNS resolver local resolution failed with: {}", e.what());
 

source/common/network/dns_impl.cc

@@ -10,6 +10,7 @@
 
 #include "common/common/assert.h"
 #include "common/common/fmt.h"
+#include "common/common/thread.h"
 #include "common/network/address_impl.h"
 #include "common/network/utility.h"
 
@@ -168,15 +169,16 @@ void DnsResolverImpl::PendingResolution::onAresGetAddrInfoCallback(int status, i
 
   if (completed_) {
     if (!cancelled_) {
-      try {
-        callback_(resolution_status, std::move(address_list));
-      } catch (const EnvoyException& e) {
+      TRY_NEEDS_AUDIT { callback_(resolution_status, std::move(address_list)); }
+      catch (const EnvoyException& e) {
         ENVOY_LOG(critical, "EnvoyException in c-ares callback: {}", e.what());
         dispatcher_.post([s = std::string(e.what())] { throw EnvoyException(s); });
-      } catch (const std::exception& e) {
+      }
+      catch (const std::exception& e) {
         ENVOY_LOG(critical, "std::exception in c-ares callback: {}", e.what());
         dispatcher_.post([s = std::string(e.what())] { throw EnvoyException(s); });
-      } catch (...) {
+      }
+      catch (...) {
         ENVOY_LOG(critical, "Unknown exception in c-ares callback");
         dispatcher_.post([] { throw EnvoyException("unknown"); });
       }

source/common/network/io_socket_handle_impl.cc

@@ -277,7 +277,7 @@ Api::IoCallUint64Result IoSocketHandleImpl::sendmsg(const Buffer::RawSlice* slic
 
 Address::InstanceConstSharedPtr getAddressFromSockAddrOrDie(const sockaddr_storage& ss,
                                                             socklen_t ss_len, os_fd_t fd) {
-  try {
+  TRY_NEEDS_AUDIT {
     // Set v6only to false so that mapped-v6 address can be normalize to v4
     // address. Though dual stack may be disabled, it's still okay to assume the
     // address is from a dual stack socket. This is because mapped-v6 address
@@ -287,7 +287,8 @@ Address::InstanceConstSharedPtr getAddressFromSockAddrOrDie(const sockaddr_stora
     // regarded as a v6 address from dual stack socket. However, this address is not going to be
     // used to create socket. Wrong knowledge of dual stack support won't hurt.
     return Address::addressFromSockAddr(ss, ss_len, /*v6only=*/false);
-  } catch (const EnvoyException& e) {
+  }
+  catch (const EnvoyException& e) {
     PANIC(fmt::format("Invalid address for fd: {}, error: {}", fd, e.what()));
   }
 }