tls: allow nullptr as slot value

include/envoy/common/BUILD

@@ -13,6 +13,7 @@ envoy_basic_cc_library(
     name = "base_includes",
     hdrs = [
         "exception.h",
+        "optref.h",
         "platform.h",
         "pure.h",
     ],

include/envoy/common/optref.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include "absl/types/optional.h"
+
+namespace Envoy {
+
+// Helper class to make it easier to work with optional references, allowing:
+//   foo(OptRef<T> t) {
+//     if (t.has_value()) {
+//       t->method();
+//     }
+//   }
+//
+// Using absl::optional directly you must write t.get()->method() which is a bit
+// more awkward.
+template <class T> struct OptRef : public absl::optional<std::reference_wrapper<T>> {
+  OptRef(T& t) : absl::optional<std::reference_wrapper<T>>(t) {}
+  OptRef() {}
+
+  T* operator->() {
+    T& ref = **this;
+    return &ref;
+  }
+
+  const T* operator->() const {
+    const T& ref = **this;
+    return &ref;
+  }
+};
+
+} // namespace Envoy

include/envoy/thread_local/thread_local.h

@@ -4,6 +4,7 @@
 #include <functional>
 #include <memory>
 
+#include "envoy/common/optref.h"
 #include "envoy/common/pure.h"
 #include "envoy/event/dispatcher.h"
 
@@ -93,8 +94,6 @@ class Slot {
   // Callers must use the TypedSlot API, below.
   virtual void runOnAllThreads(const UpdateCb& update_cb) PURE;
   virtual void runOnAllThreads(const UpdateCb& update_cb, const Event::PostCb& complete_cb) PURE;
-  virtual void runOnAllThreads(const Event::PostCb& cb) PURE;
-  virtual void runOnAllThreads(const Event::PostCb& cb, const Event::PostCb& complete_cb) PURE;
 };
 
 using SlotPtr = std::unique_ptr<Slot>;
@@ -157,39 +156,41 @@ template <class T> class TypedSlot {
   void set(InitializeCb cb) { slot_->set(cb); }
 
   /**
-   * @return a reference to the thread local object.
+   * @return an optional reference to the thread local object.
    */
-  T& get() { return slot_->getTyped<T>(); }
-  const T& get() const { return slot_->getTyped<T>(); }
+  OptRef<T> get() { return getOpt(slot_->get()); }
+  const OptRef<T> get() const { return getOpt(slot_->get()); }
 
   /**
    * @return a pointer to the thread local object.
    */
-  T* operator->() { return &get(); }
-  const T* operator->() const { return &get(); }
+  T* operator->() { return &(slot_->getTyped<T>()); }
+  const T* operator->() const { return &(slot_->getTyped<T>()); }
 
   /**
-   * UpdateCb is passed a mutable reference to the current stored data.
+   * UpdateCb is passed a mutable pointer to the current stored data. Callers
+   * can assume it's non-null if they have called set() prior to
+   * runOnAllThreads().
    *
    * NOTE: The update callback is not supposed to capture the TypedSlot, or its owner, as the owner
    * may be destructed in main thread before the update_cb gets called in a worker thread.
    */
-  using UpdateCb = std::function<void(T& obj)>;
+  using UpdateCb = std::function<void(OptRef<T> obj)>;
   void runOnAllThreads(const UpdateCb& cb) { slot_->runOnAllThreads(makeSlotUpdateCb(cb)); }
   void runOnAllThreads(const UpdateCb& cb, const Event::PostCb& complete_cb) {
     slot_->runOnAllThreads(makeSlotUpdateCb(cb), complete_cb);
   }
-  void runOnAllThreads(const Event::PostCb& cb) { slot_->runOnAllThreads(cb); }
-  void runOnAllThreads(const Event::PostCb& cb, const Event::PostCb& complete_cb) {
-    slot_->runOnAllThreads(cb, complete_cb);
-  }
 
 private:
+  static OptRef<T> getOpt(ThreadLocalObjectSharedPtr obj) {
+    if (obj) {
+      return OptRef<T>(obj->asType<T>());
+    }
+    return OptRef<T>();
+  }
+
   Slot::UpdateCb makeSlotUpdateCb(UpdateCb cb) {
-    return [cb](ThreadLocalObjectSharedPtr obj) -> ThreadLocalObjectSharedPtr {
-      cb(obj->asType<T>());
-      return obj;
-    };
+    return [cb](ThreadLocalObjectSharedPtr obj) { cb(getOpt(obj)); };
   }
 
   const SlotPtr slot_;

source/common/config/config_provider_impl.cc

@@ -25,8 +25,8 @@ ConfigSubscriptionCommonBase::~ConfigSubscriptionCommonBase() {
 }
 
 void ConfigSubscriptionCommonBase::applyConfigUpdate(const ConfigUpdateCb& update_fn) {
-  tls_.runOnAllThreads([update_fn](ThreadLocalConfig& thread_local_config) {
-    thread_local_config.config_ = update_fn(thread_local_config.config_);
+  tls_.runOnAllThreads([update_fn](OptRef<ThreadLocalConfig> thread_local_config) {
+    thread_local_config->config_ = update_fn(thread_local_config->config_);
   });
 }
 

source/common/filter/http/filter_config_discovery_impl.cc

@@ -37,7 +37,7 @@ DynamicFilterConfigProviderImpl::~DynamicFilterConfigProviderImpl() {
 const std::string& DynamicFilterConfigProviderImpl::name() { return subscription_->name(); }
 
 absl::optional<Envoy::Http::FilterFactoryCb> DynamicFilterConfigProviderImpl::config() {
-  return tls_.get().config_;
+  return tls_->config_;
 }
 
 void DynamicFilterConfigProviderImpl::validateConfig(
@@ -53,8 +53,8 @@ void DynamicFilterConfigProviderImpl::onConfigUpdate(Envoy::Http::FilterFactoryC
                                                      const std::string&,
                                                      Config::ConfigAppliedCb cb) {
   tls_.runOnAllThreads(
-      [config, cb](ThreadLocalConfig& tls) {
-        tls.config_ = config;
+      [config, cb](OptRef<ThreadLocalConfig> tls) {
+        tls->config_ = config;
         if (cb) {
           cb();
         }

source/common/router/rds_impl.cc

@@ -255,7 +255,7 @@ Router::ConfigConstSharedPtr RdsRouteConfigProviderImpl::config() { return tls_-
 void RdsRouteConfigProviderImpl::onConfigUpdate() {
   ConfigConstSharedPtr new_config(new ConfigImpl(config_update_info_->routeConfiguration(),
                                                  factory_context_, validator_, false));
-  tls_.runOnAllThreads([new_config](ThreadLocalConfig& tls) { tls.config_ = new_config; });
+  tls_.runOnAllThreads([new_config](OptRef<ThreadLocalConfig> tls) { tls->config_ = new_config; });
 
   const auto aliases = config_update_info_->resourceIdsInLastVhdsUpdate();
   // Regular (non-VHDS) RDS updates don't populate aliases fields in resources.

source/common/stats/thread_local_store.cc

@@ -205,8 +205,8 @@ void ThreadLocalStoreImpl::mergeHistograms(PostMergeCb merge_complete_cb) {
     ASSERT(!merge_in_progress_);
     merge_in_progress_ = true;
     tls_cache_->runOnAllThreads(
-        [](TlsCache& tls_cache) {
-          for (const auto& id_hist : tls_cache.tls_histogram_cache_) {
+        [](OptRef<TlsCache> tls_cache) {
+          for (const auto& id_hist : tls_cache->tls_histogram_cache_) {
             const TlsHistogramSharedPtr& tls_hist = id_hist.second;
             tls_hist->beginMerge();
           }
@@ -303,7 +303,7 @@ void ThreadLocalStoreImpl::clearScopeFromCaches(uint64_t scope_id,
   if (!shutting_down_) {
     // Perform a cache flush on all threads.
     tls_cache_->runOnAllThreads(
-        [scope_id](TlsCache& tls_cache) { tls_cache.eraseScope(scope_id); },
+        [scope_id](OptRef<TlsCache> tls_cache) { tls_cache->eraseScope(scope_id); },
         [central_cache]() { /* Holds onto central_cache until all tls caches are clear */ });
   }
 }
@@ -320,7 +320,7 @@ void ThreadLocalStoreImpl::clearHistogramFromCaches(uint64_t histogram_id) {
     // contains a patch that will implement batching together to clear multiple
     // histograms.
     tls_cache_->runOnAllThreads(
-        [histogram_id](TlsCache& tls_cache) { tls_cache.eraseHistogram(histogram_id); });
+        [histogram_id](OptRef<TlsCache> tls_cache) { tls_cache->eraseHistogram(histogram_id); });
   }
 }
 
@@ -489,7 +489,7 @@ Counter& ThreadLocalStoreImpl::ScopeImpl::counterFromStatNameWithTags(
   StatRefMap<Counter>* tls_cache = nullptr;
   StatNameHashSet* tls_rejected_stats = nullptr;
   if (!parent_.shutting_down_ && parent_.tls_cache_) {
-    TlsCacheEntry& entry = parent_.tls_cache_->get().insertScope(this->scope_id_);
+    TlsCacheEntry& entry = parent_.tls_cache_->get()->insertScope(this->scope_id_);
     tls_cache = &entry.counters_;
     tls_rejected_stats = &entry.rejected_stats_;
   }
@@ -541,7 +541,7 @@ Gauge& ThreadLocalStoreImpl::ScopeImpl::gaugeFromStatNameWithTags(
   StatRefMap<Gauge>* tls_cache = nullptr;
   StatNameHashSet* tls_rejected_stats = nullptr;
   if (!parent_.shutting_down_ && parent_.tls_cache_) {
-    TlsCacheEntry& entry = parent_.tls_cache_->get().scope_cache_[this->scope_id_];
+    TlsCacheEntry& entry = parent_.tls_cache_->get()->scope_cache_[this->scope_id_];
     tls_cache = &entry.gauges_;
     tls_rejected_stats = &entry.rejected_stats_;
   }
@@ -579,7 +579,7 @@ Histogram& ThreadLocalStoreImpl::ScopeImpl::histogramFromStatNameWithTags(
   StatNameHashMap<ParentHistogramSharedPtr>* tls_cache = nullptr;
   StatNameHashSet* tls_rejected_stats = nullptr;
   if (!parent_.shutting_down_ && parent_.tls_cache_) {
-    TlsCacheEntry& entry = parent_.tls_cache_->get().scope_cache_[this->scope_id_];
+    TlsCacheEntry& entry = parent_.tls_cache_->get()->scope_cache_[this->scope_id_];
     tls_cache = &entry.parent_histograms_;
     auto iter = tls_cache->find(final_stat_name);
     if (iter != tls_cache->end()) {
@@ -657,7 +657,7 @@ TextReadout& ThreadLocalStoreImpl::ScopeImpl::textReadoutFromStatNameWithTags(
   StatRefMap<TextReadout>* tls_cache = nullptr;
   StatNameHashSet* tls_rejected_stats = nullptr;
   if (!parent_.shutting_down_ && parent_.tls_cache_) {
-    TlsCacheEntry& entry = parent_.tls_cache_->get().insertScope(this->scope_id_);
+    TlsCacheEntry& entry = parent_.tls_cache_->get()->insertScope(this->scope_id_);
     tls_cache = &entry.text_readouts_;
     tls_rejected_stats = &entry.rejected_stats_;
   }
@@ -703,7 +703,7 @@ Histogram& ThreadLocalStoreImpl::tlsHistogram(ParentHistogramImpl& parent, uint6
 
   TlsHistogramSharedPtr* tls_histogram = nullptr;
   if (!shutting_down_ && tls_cache_) {
-    tls_histogram = &tls_cache_->get().tls_histogram_cache_[id];
+    tls_histogram = &tls_cache_->get()->tls_histogram_cache_[id];
     if (*tls_histogram != nullptr) {
       return **tls_histogram;
     }

source/common/thread_local/thread_local_impl.cc

@@ -90,15 +90,6 @@ void InstanceImpl::SlotImpl::runOnAllThreads(const UpdateCb& cb) {
   parent_.runOnAllThreads(dataCallback(cb));
 }
 
-void InstanceImpl::SlotImpl::runOnAllThreads(const Event::PostCb& cb,
-                                             const Event::PostCb& complete_cb) {
-  parent_.runOnAllThreads(wrapCallback(cb), complete_cb);
-}
-
-void InstanceImpl::SlotImpl::runOnAllThreads(const Event::PostCb& cb) {
-  parent_.runOnAllThreads(wrapCallback(cb));
-}
-
 void InstanceImpl::SlotImpl::set(InitializeCb cb) {
   ASSERT(std::this_thread::get_id() == parent_.main_thread_id_);
   ASSERT(!parent_.shutdown_);

source/common/thread_local/thread_local_impl.h

@@ -45,8 +45,6 @@ class InstanceImpl : Logger::Loggable<Logger::Id::main>, public NonCopyable, pub
     ThreadLocalObjectSharedPtr get() override;
     void runOnAllThreads(const UpdateCb& cb) override;
     void runOnAllThreads(const UpdateCb& cb, const Event::PostCb& complete_cb) override;
-    void runOnAllThreads(const Event::PostCb& cb) override;
-    void runOnAllThreads(const Event::PostCb& cb, const Event::PostCb& complete_cb) override;
     bool currentThreadRegistered() override;
     void set(InitializeCb cb) override;
 

source/common/upstream/cluster_manager_impl.cc

@@ -662,17 +662,17 @@ void ClusterManagerImpl::clusterWarmingToActive(const std::string& cluster_name)
 void ClusterManagerImpl::createOrUpdateThreadLocalCluster(ClusterData& cluster) {
   tls_.runOnAllThreads([new_cluster = cluster.cluster_->info(),
                         thread_aware_lb_factory = cluster.loadBalancerFactory()](
-                           ThreadLocalClusterManagerImpl& cluster_manager) {
-    if (cluster_manager.thread_local_clusters_.count(new_cluster->name()) > 0) {
+                           OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {
+    if (cluster_manager->thread_local_clusters_.count(new_cluster->name()) > 0) {
       ENVOY_LOG(debug, "updating TLS cluster {}", new_cluster->name());
     } else {
       ENVOY_LOG(debug, "adding TLS cluster {}", new_cluster->name());
     }
 
     auto thread_local_cluster = new ThreadLocalClusterManagerImpl::ClusterEntry(
-        cluster_manager, new_cluster, thread_aware_lb_factory);
-    cluster_manager.thread_local_clusters_[new_cluster->name()].reset(thread_local_cluster);
-    for (auto& cb : cluster_manager.update_callbacks_) {
+        *cluster_manager, new_cluster, thread_aware_lb_factory);
+    cluster_manager->thread_local_clusters_[new_cluster->name()].reset(thread_local_cluster);
+    for (auto& cb : cluster_manager->update_callbacks_) {
       cb->onClusterAddOrUpdate(*thread_local_cluster);
     }
   });
@@ -688,13 +688,13 @@ bool ClusterManagerImpl::removeCluster(const std::string& cluster_name) {
     active_clusters_.erase(existing_active_cluster);
 
     ENVOY_LOG(info, "removing cluster {}", cluster_name);
-    tls_.runOnAllThreads([cluster_name](ThreadLocalClusterManagerImpl& cluster_manager) {
-      ASSERT(cluster_manager.thread_local_clusters_.count(cluster_name) == 1);
+    tls_.runOnAllThreads([cluster_name](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {
+      ASSERT(cluster_manager->thread_local_clusters_.count(cluster_name) == 1);
       ENVOY_LOG(debug, "removing TLS cluster {}", cluster_name);
-      for (auto& cb : cluster_manager.update_callbacks_) {
+      for (auto& cb : cluster_manager->update_callbacks_) {
         cb->onClusterRemoval(cluster_name);
       }
-      cluster_manager.thread_local_clusters_.erase(cluster_name);
+      cluster_manager->thread_local_clusters_.erase(cluster_name);
     });
   }
 
@@ -822,7 +822,7 @@ void ClusterManagerImpl::updateClusterCounts() {
 }
 
 ThreadLocalCluster* ClusterManagerImpl::get(absl::string_view cluster) {
-  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get();
+  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get().value();
 
   auto entry = cluster_manager.thread_local_clusters_.find(cluster);
   if (entry != cluster_manager.thread_local_clusters_.end()) {
@@ -861,7 +861,7 @@ Http::ConnectionPool::Instance*
 ClusterManagerImpl::httpConnPoolForCluster(const std::string& cluster, ResourcePriority priority,
                                            absl::optional<Http::Protocol> protocol,
                                            LoadBalancerContext* context) {
-  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get();
+  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get().value();
 
   auto entry = cluster_manager.thread_local_clusters_.find(cluster);
   if (entry == cluster_manager.thread_local_clusters_.end()) {
@@ -887,7 +887,7 @@ ClusterManagerImpl::httpConnPoolForCluster(const std::string& cluster, ResourceP
 Tcp::ConnectionPool::Instance*
 ClusterManagerImpl::tcpConnPoolForCluster(const std::string& cluster, ResourcePriority priority,
                                           LoadBalancerContext* context) {
-  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get();
+  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get().value();
 
   auto entry = cluster_manager.thread_local_clusters_.find(cluster);
   if (entry == cluster_manager.thread_local_clusters_.end()) {
@@ -913,8 +913,8 @@ ClusterManagerImpl::tcpConnPoolForCluster(const std::string& cluster, ResourcePr
 void ClusterManagerImpl::postThreadLocalDrainConnections(const Cluster& cluster,
                                                          const HostVector& hosts_removed) {
   tls_.runOnAllThreads([name = cluster.info()->name(),
-                        hosts_removed](ThreadLocalClusterManagerImpl& cluster_manager) {
-    cluster_manager.removeHosts(name, hosts_removed);
+                        hosts_removed](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {
+    cluster_manager->removeHosts(name, hosts_removed);
   });
 }
 
@@ -927,21 +927,21 @@ void ClusterManagerImpl::postThreadLocalClusterUpdate(const Cluster& cluster, ui
                         update_params = HostSetImpl::updateHostsParams(*host_set),
                         locality_weights = host_set->localityWeights(), hosts_added, hosts_removed,
                         overprovisioning_factor = host_set->overprovisioningFactor()](
-                           ThreadLocalClusterManagerImpl& cluster_manager) {
-    cluster_manager.updateClusterMembership(name, priority, update_params, locality_weights,
-                                            hosts_added, hosts_removed, overprovisioning_factor);
+                           OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {
+    cluster_manager->updateClusterMembership(name, priority, update_params, locality_weights,
+                                             hosts_added, hosts_removed, overprovisioning_factor);
   });
 }
 
 void ClusterManagerImpl::postThreadLocalHealthFailure(const HostSharedPtr& host) {
-  tls_.runOnAllThreads([host](ThreadLocalClusterManagerImpl& cluster_manager) {
-    cluster_manager.onHostHealthFailure(host);
+  tls_.runOnAllThreads([host](OptRef<ThreadLocalClusterManagerImpl> cluster_manager) {
+    cluster_manager->onHostHealthFailure(host);
   });
 }
 
 Host::CreateConnectionData ClusterManagerImpl::tcpConnForCluster(const std::string& cluster,
                                                                  LoadBalancerContext* context) {
-  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get();
+  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get().value();
 
   auto entry = cluster_manager.thread_local_clusters_.find(cluster);
   if (entry == cluster_manager.thread_local_clusters_.end()) {
@@ -969,7 +969,7 @@ Host::CreateConnectionData ClusterManagerImpl::tcpConnForCluster(const std::stri
 }
 
 Http::AsyncClient& ClusterManagerImpl::httpAsyncClientForCluster(const std::string& cluster) {
-  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get();
+  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get().value();
   auto entry = cluster_manager.thread_local_clusters_.find(cluster);
   if (entry != cluster_manager.thread_local_clusters_.end()) {
     return entry->second->http_async_client_;
@@ -980,7 +980,7 @@ Http::AsyncClient& ClusterManagerImpl::httpAsyncClientForCluster(const std::stri
 
 ClusterUpdateCallbacksHandlePtr
 ClusterManagerImpl::addThreadLocalClusterUpdateCallbacks(ClusterUpdateCallbacks& cb) {
-  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get();
+  ThreadLocalClusterManagerImpl& cluster_manager = tls_.get().value();
   return std::make_unique<ClusterUpdateCallbacksHandleImpl>(cb, cluster_manager.update_callbacks_);
 }
 

source/extensions/clusters/aggregate/cluster.cc

@@ -91,7 +91,8 @@ void Cluster::startPreInit() {
 void Cluster::refresh(const std::function<bool(const std::string&)>& skip_predicate) {
   // Post the priority set to worker threads.
   // TODO(mattklein123): Remove "this" capture.
-  tls_.runOnAllThreads([this, skip_predicate, cluster_name = this->info()->name()]() {
+  tls_.runOnAllThreads([this, skip_predicate, cluster_name = this->info()->name()](
+                           OptRef<ThreadLocal::ThreadLocalObject>) {
     PriorityContextPtr priority_context = linearizePrioritySet(skip_predicate);
     Upstream::ThreadLocalCluster* cluster = cluster_manager_.get(cluster_name);
     ASSERT(cluster != nullptr);