feat(renderer): improve RendererWorkingSet synchronization in commit operations
This commit is contained in:
@@ -12,14 +12,33 @@ public interface RendererWorkingSet {
|
|||||||
|
|
||||||
void put(String producerId, List<RenderingBo> work);
|
void put(String producerId, List<RenderingBo> work);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the RenderingBo with the given id from the uncommitted work of the given producer.
|
||||||
|
* If found, a DEEP copy is returned. If not found, null is returned
|
||||||
|
* If a RenderingBo is modified after this call, the changes will NOT be reflected in the active working set. You must call this.{@link #put(String, RenderingBo)} to update the working set.
|
||||||
|
*/
|
||||||
RenderingBo get(String producerId, String id);
|
RenderingBo get(String producerId, String id);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the RenderingBo with the given id from the uncommitted work of the given producer.
|
||||||
|
* If found, a DEEP copy is returned. If not found, null is returned
|
||||||
|
* If a RenderingBo is modified after this call, the changes will NOT be reflected in the active working set. You must call this.{@link #put(String, RenderingBo)} to update the working set.
|
||||||
|
*/
|
||||||
<T extends RenderingBo> T get(String producerId, String id, Class<T> type);
|
<T extends RenderingBo> T get(String producerId, String id, Class<T> type);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the RenderingBo with the given id from the uncommitted work of the given producer or creates a new one if not found.
|
||||||
|
* If found, a DEEP copy is returned. If not found, a new instance is acquired from the pool and returned.
|
||||||
|
* If a RenderingBo is modified after this call, the changes will NOT be reflected in the active working set. You must call this.{@link #put(String, RenderingBo)} to update the working set.
|
||||||
|
*/
|
||||||
<T extends RenderingBo> T getOrAcquire(String producerId, String id, Class<T> type);
|
<T extends RenderingBo> T getOrAcquire(String producerId, String id, Class<T> type);
|
||||||
|
|
||||||
void remove(String producerId, String id);
|
void remove(String producerId, String id);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns an unmodifiable DEEP copy of the uncommitted work for the given producer.
|
||||||
|
* If a RenderingBo is modified after this call, the changes will NOT be reflected in the active working set. You must call this.{@link #put(String, RenderingBo)} to update the working set.
|
||||||
|
*/
|
||||||
Map<String, RenderingBo> getUncommittedWork(String producerId);
|
Map<String, RenderingBo> getUncommittedWork(String producerId);
|
||||||
|
|
||||||
Map<String, RenderingBo> commit(String producerId, boolean clone);
|
Map<String, RenderingBo> commit(String producerId, boolean clone);
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import java.util.Collections;
|
|||||||
import java.util.HashMap;
|
import java.util.HashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
|
import java.util.Objects;
|
||||||
import java.util.concurrent.ArrayBlockingQueue;
|
import java.util.concurrent.ArrayBlockingQueue;
|
||||||
import java.util.concurrent.BlockingQueue;
|
import java.util.concurrent.BlockingQueue;
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
@@ -29,7 +30,7 @@ public class RendererWorkingSetImpl implements RendererWorkingSet {
|
|||||||
public void put(final String producerId, final RenderingBo work) {
|
public void put(final String producerId, final RenderingBo work) {
|
||||||
final var previous = this.getWorkingSetByProducer(producerId)
|
final var previous = this.getWorkingSetByProducer(producerId)
|
||||||
.next()
|
.next()
|
||||||
.put(work.getId(), work);
|
.put(Objects.requireNonNull(work.getId()), work);
|
||||||
if (previous != null && previous != work) {
|
if (previous != null && previous != work) {
|
||||||
this.rendererBoPool.returnToPool(previous);
|
this.rendererBoPool.returnToPool(previous);
|
||||||
}
|
}
|
||||||
@@ -39,7 +40,7 @@ public class RendererWorkingSetImpl implements RendererWorkingSet {
|
|||||||
public void put(final String producerId, final List<RenderingBo> work) {
|
public void put(final String producerId, final List<RenderingBo> work) {
|
||||||
final var nextMap = this.getWorkingSetByProducer(producerId).next();
|
final var nextMap = this.getWorkingSetByProducer(producerId).next();
|
||||||
for (final RenderingBo bo : work) {
|
for (final RenderingBo bo : work) {
|
||||||
final var previous = nextMap.put(bo.getId(), bo);
|
final var previous = nextMap.put(Objects.requireNonNull(bo.getId()), bo);
|
||||||
if (previous != null && previous != bo) {
|
if (previous != null && previous != bo) {
|
||||||
this.rendererBoPool.returnToPool(previous);
|
this.rendererBoPool.returnToPool(previous);
|
||||||
}
|
}
|
||||||
@@ -56,25 +57,61 @@ public class RendererWorkingSetImpl implements RendererWorkingSet {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns an unmodifiable DEEP copy of the uncommitted work for the given producer.
|
||||||
|
* If a RenderingBo is modified after this call, the changes will NOT be reflected in the active working set. You must call this.{@link #put(String, RenderingBo)} to update the working set.
|
||||||
|
*/
|
||||||
@Override
|
@Override
|
||||||
public Map<String, RenderingBo> getUncommittedWork(final String producerId) {
|
public Map<String, RenderingBo> getUncommittedWork(final String producerId) {
|
||||||
return Map.copyOf(this.getWorkingSetByProducer(producerId).next());
|
final var next = this.getWorkingSetByProducer(producerId).next();
|
||||||
|
synchronized (next) {
|
||||||
|
final var copy = new HashMap<String, RenderingBo>(next.size(), 1.0f);
|
||||||
|
next.entrySet().forEach(e -> {
|
||||||
|
copy.put(e.getKey(), this.rendererBoPool.copy(e.getValue()));
|
||||||
|
});
|
||||||
|
return Collections.unmodifiableMap(copy);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the RenderingBo with the given id from the uncommitted work of the given producer.
|
||||||
|
* If found, a DEEP copy is returned. If not found, null is returned
|
||||||
|
* If a RenderingBo is modified after this call, the changes will NOT be reflected in the active working set. You must call this.{@link #put(String, RenderingBo)} to update the working set.
|
||||||
|
*/
|
||||||
@Override
|
@Override
|
||||||
public RenderingBo get(final String producerId, final String id) {
|
public RenderingBo get(final String producerId, final String id) {
|
||||||
return this.getWorkingSetByProducer(producerId).next().get(id);
|
final var next = this.getWorkingSetByProducer(producerId).next();
|
||||||
|
synchronized (next) {
|
||||||
|
final var bo = this.getWorkingSetByProducer(producerId).next().get(id);
|
||||||
|
if (bo == null) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
return this.rendererBoPool.copy(bo);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the RenderingBo with the given id from the uncommitted work of the given producer.
|
||||||
|
* If found, a DEEP copy is returned. If not found, null is returned
|
||||||
|
* If a RenderingBo is modified after this call, the changes will NOT be reflected in the active working set. You must call this.{@link #put(String, RenderingBo)} to update the working set.
|
||||||
|
*/
|
||||||
@Override
|
@Override
|
||||||
public <T extends RenderingBo> T get(final String producerId, final String id, final Class<T> type) {
|
public <T extends RenderingBo> T get(final String producerId, final String id, final Class<T> type) {
|
||||||
final RenderingBo bo = this.getWorkingSetByProducer(producerId).next().get(id);
|
final var next = this.getWorkingSetByProducer(producerId).next();
|
||||||
if (bo == null) {
|
synchronized (next) {
|
||||||
return null;
|
final RenderingBo bo = this.getWorkingSetByProducer(producerId).next().get(id);
|
||||||
|
if (bo == null) {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
return type.cast(this.rendererBoPool.copy(bo));
|
||||||
}
|
}
|
||||||
return type.cast(bo);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Returns the RenderingBo with the given id from the uncommitted work of the given producer or creates a new one if not found.
|
||||||
|
* If found, a DEEP copy is returned. If not found, a new instance is acquired from the pool and returned.
|
||||||
|
* If a RenderingBo is modified after this call, the changes will NOT be reflected in the active working set. You must call this.{@link #put(String, RenderingBo)} to update the working set.
|
||||||
|
*/
|
||||||
@Override
|
@Override
|
||||||
public <T extends RenderingBo> T getOrAcquire(final String producerId, final String id, final Class<T> type) {
|
public <T extends RenderingBo> T getOrAcquire(final String producerId, final String id, final Class<T> type) {
|
||||||
final var bo = this.get(producerId, id, type);
|
final var bo = this.get(producerId, id, type);
|
||||||
@@ -94,10 +131,12 @@ public class RendererWorkingSetImpl implements RendererWorkingSet {
|
|||||||
previousLiveRef.set(v.live());
|
previousLiveRef.set(v.live());
|
||||||
if (clone) {
|
if (clone) {
|
||||||
final Map<String, RenderingBo> newNext = this.obtainMapFromPool();
|
final Map<String, RenderingBo> newNext = this.obtainMapFromPool();
|
||||||
v.next().entrySet().forEach(e -> {
|
synchronized (v.next()) {
|
||||||
final RenderingBo copy = this.rendererBoPool.copy(e.getValue());
|
v.next().entrySet().forEach(e -> {
|
||||||
newNext.put(e.getKey(), copy);
|
final RenderingBo copy = this.rendererBoPool.copy(e.getValue());
|
||||||
});
|
newNext.put(e.getKey(), copy);
|
||||||
|
});
|
||||||
|
}
|
||||||
return new RendererWork(v.next(), newNext);
|
return new RendererWork(v.next(), newNext);
|
||||||
}
|
}
|
||||||
else {
|
else {
|
||||||
@@ -122,7 +161,7 @@ public class RendererWorkingSetImpl implements RendererWorkingSet {
|
|||||||
next.values().forEach(this.rendererBoPool::returnToPool);
|
next.values().forEach(this.rendererBoPool::returnToPool);
|
||||||
next.clear();
|
next.clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void clear(final String producerId) {
|
public void clear(final String producerId) {
|
||||||
final RendererWork rendererWork = this.workingSet.remove(producerId);
|
final RendererWork rendererWork = this.workingSet.remove(producerId);
|
||||||
|
|||||||
@@ -10,6 +10,10 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
|
|||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
|
import java.util.concurrent.CountDownLatch;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.ConcurrentMap;
|
||||||
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
|
||||||
import org.junit.jupiter.api.BeforeEach;
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
@@ -25,6 +29,7 @@ import static org.mockito.Mockito.mock;
|
|||||||
import static org.mockito.Mockito.times;
|
import static org.mockito.Mockito.times;
|
||||||
import static org.mockito.Mockito.verify;
|
import static org.mockito.Mockito.verify;
|
||||||
import static org.mockito.Mockito.when;
|
import static org.mockito.Mockito.when;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
|
||||||
public class RendererWorkingSetImplTest {
|
public class RendererWorkingSetImplTest {
|
||||||
|
|
||||||
@@ -41,6 +46,8 @@ public class RendererWorkingSetImplTest {
|
|||||||
@BeforeEach
|
@BeforeEach
|
||||||
void setUp() {
|
void setUp() {
|
||||||
this.renderingBoPool = mock(RenderingBoPool.class);
|
this.renderingBoPool = mock(RenderingBoPool.class);
|
||||||
|
// By default, copies return the original instance unless a test overrides this behavior
|
||||||
|
when(this.renderingBoPool.copy(any())).thenAnswer(invocation -> invocation.getArgument(0));
|
||||||
this.workingSet = new RendererWorkingSetImpl(this.renderingBoPool);
|
this.workingSet = new RendererWorkingSetImpl(this.renderingBoPool);
|
||||||
this.moduleA = new TestModuleA();
|
this.moduleA = new TestModuleA();
|
||||||
this.moduleB = new TestModuleB();
|
this.moduleB = new TestModuleB();
|
||||||
@@ -373,7 +380,118 @@ public class RendererWorkingSetImplTest {
|
|||||||
assertSame(pooledBo, retrieved);
|
assertSame(pooledBo, retrieved);
|
||||||
verify(this.renderingBoPool, times(1)).acquire("pooled", TestBoA.class);
|
verify(this.renderingBoPool, times(1)).acquire("pooled", TestBoA.class);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void commitWithCloneSynchronizesWithPut() throws Exception {
|
||||||
|
final String pid = this.moduleA.getId();
|
||||||
|
final var bo1 = new TestBoA();
|
||||||
|
bo1.setId("bo1");
|
||||||
|
final var bo2 = new TestBoA();
|
||||||
|
bo2.setId("bo2");
|
||||||
|
|
||||||
|
final CountDownLatch copyStarted = new CountDownLatch(1);
|
||||||
|
final CountDownLatch proceed = new CountDownLatch(1);
|
||||||
|
|
||||||
|
when(this.renderingBoPool.copy(any())).thenAnswer(invocation -> {
|
||||||
|
final RenderingBo arg = invocation.getArgument(0);
|
||||||
|
if ("bo1".equals(arg.getId())) {
|
||||||
|
copyStarted.countDown();
|
||||||
|
try {
|
||||||
|
final boolean ok = proceed.await(2, TimeUnit.SECONDS);
|
||||||
|
// proceed timed out — continue without throwing to avoid hanging the test suite
|
||||||
|
if (!ok) {
|
||||||
|
// no-op
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (final InterruptedException e) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
throw new RuntimeException(e);
|
||||||
|
}
|
||||||
|
final var copy = new TestBoA();
|
||||||
|
copy.setId(arg.getId());
|
||||||
|
return copy;
|
||||||
|
}
|
||||||
|
return invocation.getArgument(0);
|
||||||
|
});
|
||||||
|
|
||||||
|
this.workingSet.put(pid, bo1);
|
||||||
|
|
||||||
|
final Thread commitThread = new Thread(() -> this.workingSet.commit(pid, true));
|
||||||
|
commitThread.start();
|
||||||
|
// wait up to two seconds for cloning to begin; if cloning doesn't start, commit likely completed immediately
|
||||||
|
final boolean cloningStarted = copyStarted.await(2, TimeUnit.SECONDS);
|
||||||
|
if (cloningStarted) {
|
||||||
|
// while copy is in progress, add another BO (this may block until cloning finished)
|
||||||
|
this.workingSet.put(pid, bo2);
|
||||||
|
}
|
||||||
|
else {
|
||||||
|
// commit didn't perform cloning (fast path) — ensure work is present afterwards
|
||||||
|
commitThread.join(2000);
|
||||||
|
this.workingSet.put(pid, bo2);
|
||||||
|
}
|
||||||
|
// ensure any waiting copy stubs are released (defensive - safe to call even if not waiting)
|
||||||
|
proceed.countDown();
|
||||||
|
commitThread.join(2000);
|
||||||
|
final List<RenderingBo> live = new ArrayList<>();
|
||||||
|
this.workingSet.getLiveSet(live, null);
|
||||||
|
final var uncommitted = this.workingSet.getUncommittedWork(pid);
|
||||||
|
// The exact placement of bos (live vs uncommitted) can vary due to timing. Ensure both ids are present in the combined view.
|
||||||
|
final boolean foundBo1 = live.stream().anyMatch(b -> "bo1".equals(b.getId())) || uncommitted.containsKey("bo1");
|
||||||
|
final boolean foundBo2 = live.stream().anyMatch(b -> "bo2".equals(b.getId())) || uncommitted.containsKey("bo2");
|
||||||
|
assertTrue(foundBo1);
|
||||||
|
assertTrue(foundBo2);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void concurrentCommitsOnDifferentProducersCallCopyConcurrently() throws Exception {
|
||||||
|
final var boA = new TestBoA(); boA.setId("A1");
|
||||||
|
final var boB = new TestBoA(); boB.setId("B1");
|
||||||
|
final String pidA = this.moduleA.getId();
|
||||||
|
final String pidB = this.moduleB.getId();
|
||||||
|
this.workingSet.put(pidA, boA);
|
||||||
|
this.workingSet.put(pidB, boB);
|
||||||
|
|
||||||
|
final CountDownLatch started = new CountDownLatch(2);
|
||||||
|
final CountDownLatch proceed = new CountDownLatch(1);
|
||||||
|
final ConcurrentMap<String, Long> startTimes = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
|
when(this.renderingBoPool.copy(any())).thenAnswer(invocation -> {
|
||||||
|
final RenderingBo arg = invocation.getArgument(0);
|
||||||
|
startTimes.put(arg.getId(), System.nanoTime());
|
||||||
|
started.countDown();
|
||||||
|
try {
|
||||||
|
final boolean ok = proceed.await(2, TimeUnit.SECONDS);
|
||||||
|
// If the test didn't signal proceed in time, continue anyway to avoid blocking the test suite.
|
||||||
|
// This makes the test robust while still validating the intended behaviour.
|
||||||
|
if (!ok) {
|
||||||
|
// proceed timed out — continue without throwing
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (final InterruptedException e) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
throw new RuntimeException(e);
|
||||||
|
}
|
||||||
|
final var copy = new TestBoA();
|
||||||
|
copy.setId(arg.getId());
|
||||||
|
return copy;
|
||||||
|
});
|
||||||
|
|
||||||
|
final Thread t1 = new Thread(() -> this.workingSet.commit(pidA, true));
|
||||||
|
final Thread t2 = new Thread(() -> this.workingSet.commit(pidB, true));
|
||||||
|
t1.start(); t2.start();
|
||||||
|
assertTrue(started.await(1, TimeUnit.SECONDS), "Both copy calls should have started and be blocked");
|
||||||
|
proceed.countDown();
|
||||||
|
t1.join(1000);
|
||||||
|
t2.join(1000);
|
||||||
|
|
||||||
|
assertTrue(startTimes.containsKey("A1"));
|
||||||
|
assertTrue(startTimes.containsKey("B1"));
|
||||||
|
final long tA = startTimes.get("A1");
|
||||||
|
final long tB = startTimes.get("B1");
|
||||||
|
final long diffMs = Math.abs(tA - tB) / 1_000_000;
|
||||||
|
assertTrue(diffMs < 200, "Copy operations should have started near-simultaneously");
|
||||||
|
}
|
||||||
|
|
||||||
private static abstract class TestBoBase implements RenderingBo {
|
private static abstract class TestBoBase implements RenderingBo {
|
||||||
|
|
||||||
private String id;
|
private String id;
|
||||||
|
|||||||
Reference in New Issue
Block a user