Fix #1632 - Occasional Segfault with LongCounter instrument

sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h

@@ -48,21 +48,31 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
     {
       auto aggr = DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_);
       aggr->Aggregate(measurement.second);
+      cumulative_hashmap_lock_.lock();
       auto prev = cumulative_hash_map_->Get(measurement.first);
+      cumulative_hashmap_lock_.unlock();
       if (prev)
       {
         auto delta = prev->Diff(*aggr);
+        cumulative_hashmap_lock_.lock();
         cumulative_hash_map_->Set(measurement.first,
                                   DefaultAggregation::CloneAggregation(
                                       aggregation_type_, instrument_descriptor_, *delta));
+        cumulative_hashmap_lock_.unlock();
+        delta_hashmap_lock_.lock();
         delta_hash_map_->Set(measurement.first, std::move(delta));
+        delta_hashmap_lock_.unlock();
       }
       else
       {
+        cumulative_hashmap_lock_.lock();
         cumulative_hash_map_->Set(
             measurement.first,
             DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, *aggr));
+        cumulative_hashmap_lock_.unlock();
+        delta_hashmap_lock_.lock();
         delta_hash_map_->Set(measurement.first, std::move(aggr));
+        delta_hashmap_lock_.unlock();
       }
     }
   }
@@ -96,10 +106,16 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
                nostd::function_ref<bool(MetricData)> metric_collection_callback) noexcept override
   {
 
-    auto status = temporal_metric_storage_.buildMetrics(collector, collectors, sdk_start_ts,
-                                                        collection_ts, std::move(delta_hash_map_),
-                                                        metric_collection_callback);
-    delta_hash_map_.reset(new AttributesHashMap());
+    std::shared_ptr<AttributesHashMap> delta_metrics = nullptr;
+    {
+      std::lock_guard<opentelemetry::common::SpinLockMutex> guard(delta_hashmap_lock_);
+      delta_metrics = std::move(delta_hash_map_);
+      delta_hash_map_.reset(new AttributesHashMap);
+    }
+
+    auto status =
+        temporal_metric_storage_.buildMetrics(collector, collectors, sdk_start_ts, collection_ts,
+                                              delta_metrics, metric_collection_callback);
     return status;
   }
 
@@ -109,7 +125,10 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
   const AttributesProcessor *attributes_processor_;
   void *state_;
   std::unique_ptr<AttributesHashMap> cumulative_hash_map_;
+  opentelemetry::common::SpinLockMutex cumulative_hashmap_lock_;
   std::unique_ptr<AttributesHashMap> delta_hash_map_;
+  opentelemetry::common::SpinLockMutex delta_hashmap_lock_;
+
   TemporalMetricStorage temporal_metric_storage_;
 };
 

sdk/include/opentelemetry/sdk/metrics/state/attributes_hashmap.h

@@ -3,7 +3,6 @@
 
 #pragma once
 #ifndef ENABLE_METRICS_PREVIEW
-#  include "opentelemetry/common/spin_lock_mutex.h"
 #  include "opentelemetry/nostd/function_ref.h"
 #  include "opentelemetry/sdk/common/attribute_utils.h"
 #  include "opentelemetry/sdk/common/attributemap_hash.h"
@@ -13,7 +12,6 @@
 
 #  include <functional>
 #  include <memory>
-#  include <mutex>
 #  include <unordered_map>
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -37,7 +35,6 @@ class AttributesHashMap
 public:
   Aggregation *Get(const MetricAttributes &attributes) const
   {
-    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
     auto it = hash_map_.find(attributes);
     if (it != hash_map_.end())
     {
@@ -52,7 +49,6 @@ class AttributesHashMap
    */
   bool Has(const MetricAttributes &attributes) const
   {
-    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
     return (hash_map_.find(attributes) == hash_map_.end()) ? false : true;
   }
 
@@ -64,8 +60,6 @@ class AttributesHashMap
   Aggregation *GetOrSetDefault(const MetricAttributes &attributes,
                                std::function<std::unique_ptr<Aggregation>()> aggregation_callback)
   {
-    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
-
     auto it = hash_map_.find(attributes);
     if (it != hash_map_.end())
     {
@@ -81,7 +75,6 @@ class AttributesHashMap
    */
   void Set(const MetricAttributes &attributes, std::unique_ptr<Aggregation> value)
   {
-    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
     hash_map_[attributes] = std::move(value);
   }
 
@@ -91,7 +84,6 @@ class AttributesHashMap
   bool GetAllEnteries(
       nostd::function_ref<bool(const MetricAttributes &, Aggregation &)> callback) const
   {
-    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
     for (auto &kv : hash_map_)
     {
       if (!callback(kv.first, *(kv.second.get())))
@@ -105,17 +97,11 @@ class AttributesHashMap
   /**
    * Return the size of hash.
    */
-  size_t Size()
-  {
-    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
-    return hash_map_.size();
-  }
+  size_t Size() { return hash_map_.size(); }
 
 private:
   std::unordered_map<MetricAttributes, std::unique_ptr<Aggregation>, AttributeHashGenerator>
       hash_map_;
-
-  mutable opentelemetry::common::SpinLockMutex lock_;
 };
 }  // namespace metrics
 

sdk/include/opentelemetry/sdk/metrics/state/sync_metric_storage.h

@@ -52,6 +52,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
       return;
     }
     exemplar_reservoir_->OfferMeasurement(value, {}, context, std::chrono::system_clock::now());
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
     attributes_hashmap_->GetOrSetDefault({}, create_default_aggregation_)->Aggregate(value);
   }
 
@@ -67,6 +68,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
     exemplar_reservoir_->OfferMeasurement(value, attributes, context,
                                           std::chrono::system_clock::now());
     auto attr = attributes_processor_->process(attributes);
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
     attributes_hashmap_->GetOrSetDefault(attr, create_default_aggregation_)->Aggregate(value);
   }
 
@@ -77,6 +79,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
       return;
     }
     exemplar_reservoir_->OfferMeasurement(value, {}, context, std::chrono::system_clock::now());
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
     attributes_hashmap_->GetOrSetDefault({}, create_default_aggregation_)->Aggregate(value);
   }
 
@@ -93,6 +96,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
     exemplar_reservoir_->OfferMeasurement(value, attributes, context,
                                           std::chrono::system_clock::now());
     auto attr = attributes_processor_->process(attributes);
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
     attributes_hashmap_->GetOrSetDefault(attr, create_default_aggregation_)->Aggregate(value);
   }
 
@@ -117,6 +121,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
   std::function<std::unique_ptr<Aggregation>()> create_default_aggregation_;
   nostd::shared_ptr<ExemplarReservoir> exemplar_reservoir_;
   TemporalMetricStorage temporal_metric_storage_;
+  opentelemetry::common::SpinLockMutex lock_;
 };
 
 }  // namespace metrics

sdk/src/metrics/state/sync_metric_storage.cc

@@ -20,8 +20,12 @@ bool SyncMetricStorage::Collect(CollectorHandle *collector,
   // Add the current delta metrics to `unreported metrics stash` for all the collectors,
   // this will also empty the delta metrics hashmap, and make it available for
   // recordings
-  std::shared_ptr<AttributesHashMap> delta_metrics = std::move(attributes_hashmap_);
-  attributes_hashmap_.reset(new AttributesHashMap);
+  std::shared_ptr<AttributesHashMap> delta_metrics = nullptr;
+  {
+    std::lock_guard<opentelemetry::common::SpinLockMutex> guard(lock_);
+    delta_metrics = std::move(attributes_hashmap_);
+    attributes_hashmap_.reset(new AttributesHashMap);
+  }
 
   return temporal_metric_storage_.buildMetrics(collector, collectors, sdk_start_ts, collection_ts,
                                                std::move(delta_metrics), callback);

sdk/test/metrics/attributes_hashmap_benchmark.cc

@@ -25,14 +25,17 @@ void BM_AttributseHashMap(benchmark::State &state)
   std::vector<MetricAttributes> attributes = {{{"k1", "v1"}, {"k2", "v2"}},
                                               {{"k1", "v1"}, {"k2", "v2"}, {"k3", "v3"}}};
 
-  auto work = [&attributes, &hash_map](const size_t i) {
+  std::mutex m;
+
+  auto work = [&attributes, &hash_map, &m](const size_t i) {
     std::function<std::unique_ptr<Aggregation>()> create_default_aggregation =
         []() -> std::unique_ptr<Aggregation> {
       return std::unique_ptr<Aggregation>(new DropAggregation);
     };
-
+    m.lock();
     hash_map.GetOrSetDefault(attributes[i % 2], create_default_aggregation)->Aggregate(1l);
     benchmark::DoNotOptimize(hash_map.Has(attributes[i % 2]));
+    m.unlock();
   };
   while (state.KeepRunning())
   {