Remove contention between threads and worker thread when register span names. (#729)

* Remove contention between threads and worker thread when register span names.

* Mark getRegisteredSpanNamesForCollection as test only.

3 files changed, 57 insertions, 11 deletions

diff --git a/impl_core/src/main/java/io/opencensus/implcore/trace/TraceComponentImplBase.java b/impl_core/src/main/java/io/opencensus/implcore/trace/TraceComponentImplBase.java
index f5f70a52..f4c5ef76 100644
--- a/impl_core/src/main/java/io/opencensus/implcore/trace/TraceComponentImplBase.java
+++ b/impl_core/src/main/java/io/opencensus/implcore/trace/TraceComponentImplBase.java
@@ -41,7 +41,6 @@ public final class TraceComponentImplBase {
   private final ExportComponentImpl exportComponent;
   private final PropagationComponent propagationComponent = new PropagationComponentImpl();
   private final Clock clock;
-  private final StartEndHandler startEndHandler;
   private final TraceConfig traceConfig = new TraceConfigImpl();
   private final Tracer tracer;
 
@@ -56,11 +55,11 @@ public final class TraceComponentImplBase {
     this.clock = clock;
     // TODO(bdrutu): Add a config/argument for supportInProcessStores.
     if (eventQueue instanceof SimpleEventQueue) {
-      exportComponent = ExportComponentImpl.createWithoutInProcessStores();
+      exportComponent = ExportComponentImpl.createWithoutInProcessStores(eventQueue);
     } else {
-      exportComponent = ExportComponentImpl.createWithInProcessStores();
+      exportComponent = ExportComponentImpl.createWithInProcessStores(eventQueue);
     }
-    startEndHandler =
+    StartEndHandler startEndHandler =
         new StartEndHandlerImpl(
             exportComponent.getSpanExporter(),
             exportComponent.getRunningSpanStore(),
diff --git a/impl_core/src/main/java/io/opencensus/implcore/trace/export/ExportComponentImpl.java b/impl_core/src/main/java/io/opencensus/implcore/trace/export/ExportComponentImpl.java
index 7d9b1895..c30f6b3c 100644
--- a/impl_core/src/main/java/io/opencensus/implcore/trace/export/ExportComponentImpl.java
+++ b/impl_core/src/main/java/io/opencensus/implcore/trace/export/ExportComponentImpl.java
@@ -16,6 +16,7 @@
 
 package io.opencensus.implcore.trace.export;
 
+import io.opencensus.implcore.internal.EventQueue;
 import io.opencensus.trace.export.ExportComponent;
 import io.opencensus.trace.export.RunningSpanStore;
 import io.opencensus.trace.export.SampledSpanStore;
@@ -54,8 +55,8 @@ public final class ExportComponentImpl extends ExportComponent {
    *
    * @return a new {@code ExportComponentImpl}.
    */
-  public static ExportComponentImpl createWithInProcessStores() {
-    return new ExportComponentImpl(true);
+  public static ExportComponentImpl createWithInProcessStores(EventQueue eventQueue) {
+    return new ExportComponentImpl(true, eventQueue);
   }
 
   /**
@@ -64,8 +65,8 @@ public final class ExportComponentImpl extends ExportComponent {
    *
    * @return a new {@code ExportComponentImpl}.
    */
-  public static ExportComponentImpl createWithoutInProcessStores() {
-    return new ExportComponentImpl(false);
+  public static ExportComponentImpl createWithoutInProcessStores(EventQueue eventQueue) {
+    return new ExportComponentImpl(false, eventQueue);
   }
 
   /**
@@ -74,9 +75,9 @@ public final class ExportComponentImpl extends ExportComponent {
    * @param supportInProcessStores {@code true} to instantiate {@link RunningSpanStore} and {@link
    *     SampledSpanStore}.
    */
-  private ExportComponentImpl(boolean supportInProcessStores) {
+  private ExportComponentImpl(boolean supportInProcessStores, EventQueue eventQueue) {
     this.spanExporter = SpanExporterImpl.create(EXPORTER_BUFFER_SIZE, EXPORTER_SCHEDULE_DELAY_MS);
     this.runningSpanStore = supportInProcessStores ? new RunningSpanStoreImpl() : null;
-    this.sampledSpanStore = supportInProcessStores ? new SampledSpanStoreImpl() : null;
+    this.sampledSpanStore = supportInProcessStores ? new SampledSpanStoreImpl(eventQueue) : null;
   }
 }
diff --git a/impl_core/src/main/java/io/opencensus/implcore/trace/export/SampledSpanStoreImpl.java b/impl_core/src/main/java/io/opencensus/implcore/trace/export/SampledSpanStoreImpl.java
index 126d4634..9235d4f1 100644
--- a/impl_core/src/main/java/io/opencensus/implcore/trace/export/SampledSpanStoreImpl.java
+++ b/impl_core/src/main/java/io/opencensus/implcore/trace/export/SampledSpanStoreImpl.java
@@ -17,6 +17,7 @@
 package io.opencensus.implcore.trace.export;
 
 import com.google.common.collect.EvictingQueue;
+import io.opencensus.implcore.internal.EventQueue;
 import io.opencensus.implcore.trace.SpanImpl;
 import io.opencensus.trace.Status;
 import io.opencensus.trace.Status.CanonicalCode;
@@ -48,6 +49,10 @@ public final class SampledSpanStoreImpl extends SampledSpanStore {
       NUM_SAMPLES_PER_LATENCY_BUCKET * NUM_LATENCY_BUCKETS
           + NUM_SAMPLES_PER_ERROR_BUCKET * NUM_ERROR_BUCKETS;
 
+  // Used to stream the register/unregister events to the implementation to avoid lock contention
+  // between the main threads and the worker thread.
+  private final EventQueue eventQueue;
+
   @GuardedBy("samples")
   private final Map<String, PerSpanNameSamples> samples;
 
@@ -229,8 +234,9 @@ public final class SampledSpanStoreImpl extends SampledSpanStore {
   }
 
   /** Constructs a new {@code SampledSpanStoreImpl}. */
-  SampledSpanStoreImpl() {
+  SampledSpanStoreImpl(EventQueue eventQueue) {
     samples = new HashMap<String, PerSpanNameSamples>();
+    this.eventQueue = eventQueue;
   }
 
   @Override
@@ -269,6 +275,10 @@ public final class SampledSpanStoreImpl extends SampledSpanStore {
 
   @Override
   public void registerSpanNamesForCollection(Collection<String> spanNames) {
+    eventQueue.enqueue(new RegisterSpanNameEvent(this, spanNames));
+  }
+
+  private void internaltRegisterSpanNamesForCollection(Collection<String> spanNames) {
     synchronized (samples) {
       for (String spanName : spanNames) {
         if (!samples.containsKey(spanName)) {
@@ -278,13 +288,49 @@ public final class SampledSpanStoreImpl extends SampledSpanStore {
     }
   }
 
+  private static final class RegisterSpanNameEvent implements EventQueue.Entry {
+    private final SampledSpanStoreImpl sampledSpanStore;
+    private final Collection<String> spanNames;
+
+    private RegisterSpanNameEvent(
+        SampledSpanStoreImpl sampledSpanStore, Collection<String> spanNames) {
+      this.sampledSpanStore = sampledSpanStore;
+      this.spanNames = new ArrayList<String>(spanNames);
+    }
+
+    @Override
+    public void process() {
+      sampledSpanStore.internaltRegisterSpanNamesForCollection(spanNames);
+    }
+  }
+
   @Override
   public void unregisterSpanNamesForCollection(Collection<String> spanNames) {
+    eventQueue.enqueue(new UnregisterSpanNameEvent(this, spanNames));
+  }
+
+  private void internalUnregisterSpanNamesForCollection(Collection<String> spanNames) {
     synchronized (samples) {
       samples.keySet().removeAll(spanNames);
     }
   }
 
+  private static final class UnregisterSpanNameEvent implements EventQueue.Entry {
+    private final SampledSpanStoreImpl sampledSpanStore;
+    private final Collection<String> spanNames;
+
+    private UnregisterSpanNameEvent(
+        SampledSpanStoreImpl sampledSpanStore, Collection<String> spanNames) {
+      this.sampledSpanStore = sampledSpanStore;
+      this.spanNames = new ArrayList<String>(spanNames);
+    }
+
+    @Override
+    public void process() {
+      sampledSpanStore.internalUnregisterSpanNamesForCollection(spanNames);
+    }
+  }
+
   @Override
   public Set<String> getRegisteredSpanNamesForCollection() {
     synchronized (samples) {