Skip to content

Commit

Permalink
Fix concurrent modification exception in ComponentRegistry (open-tele…
Browse files Browse the repository at this point in the history
…metry#4951)

* Fix concurrent modification exception in ComponentRegistry

* Reduce number of threads and iterations
  • Loading branch information
jack-berg authored and dmarkwat committed Dec 30, 2022
1 parent a1a87fa commit e867d28
Show file tree
Hide file tree
Showing 2 changed files with 51 additions and 2 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,9 @@
package io.opentelemetry.sdk.internal;

import io.opentelemetry.api.common.Attributes;
import io.opentelemetry.api.internal.GuardedBy;
import io.opentelemetry.sdk.common.InstrumentationScopeInfo;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.IdentityHashMap;
Expand Down Expand Up @@ -44,6 +46,9 @@ public final class ComponentRegistry<V> {
private final Map<String, Map<String, Map<String, V>>> componentByNameVersionAndSchema =
new ConcurrentHashMap<>();

private final Object lock = new Object();

@GuardedBy("lock")
private final Set<V> allComponents = Collections.newSetFromMap(new IdentityHashMap<>());

private final Function<InstrumentationScopeInfo, V> factory;
Expand Down Expand Up @@ -109,7 +114,9 @@ public V get(

private V buildComponent(InstrumentationScopeInfo instrumentationScopeInfo) {
V component = factory.apply(instrumentationScopeInfo);
allComponents.add(component);
synchronized (lock) {
allComponents.add(component);
}
return component;
}

Expand All @@ -119,6 +126,8 @@ private V buildComponent(InstrumentationScopeInfo instrumentationScopeInfo) {
* @return a {@code Collection} view of the registered components.
*/
public Collection<V> getComponents() {
return Collections.unmodifiableCollection(allComponents);
synchronized (lock) {
return Collections.unmodifiableCollection(new ArrayList<>(allComponents));
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,18 @@

package io.opentelemetry.sdk.internal;

import static java.util.stream.Collectors.joining;
import static org.assertj.core.api.Assertions.assertThat;

import io.opentelemetry.api.common.Attributes;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.IntStream;
import org.junit.jupiter.api.Test;

class ComponentRegistryTest {
Expand Down Expand Up @@ -52,5 +61,36 @@ void get_DifferentInstance() {
.isNotSameAs(registry.get(NAME, null, null, Attributes.empty()));
}

@Test
@SuppressWarnings("ReturnValueIgnored")
void getComponents_HighConcurrency() throws ExecutionException, InterruptedException {
List<Future<?>> futures = new ArrayList<>();
Random random = new Random();
int concurrency = 2;
ExecutorService executor = Executors.newFixedThreadPool(concurrency);

try {
for (int i = 0; i < 100; i++) {
futures.add(
executor.submit(
() -> {
String name =
IntStream.range(0, 20)
.mapToObj(unused -> String.valueOf((char) random.nextInt(26)))
.collect(joining());
registry.get(name, null, null, Attributes.empty());
}));
futures.add(
executor.submit(() -> registry.getComponents().forEach(TestComponent::hashCode)));
}

for (Future<?> future : futures) {
future.get();
}
} finally {
executor.shutdown();
}
}

private static final class TestComponent {}
}

0 comments on commit e867d28

Please sign in to comment.