Skip to content

Commit

Permalink
Issue-1337: Fix unstable concurrent code in PointcutLatch, fix subscr…
Browse files Browse the repository at this point in the history 
…iption tests getting latch exceptions due to missing expectations, make hapi-fhir-jpaserver-subscription tests load StructureDefinitions outside latch timers, as this can be slow on busy machines (#1338)
  • Loading branch information
@srdo @fil512
srdo authored and fil512 committed Jun 10, 2019
1 parent a952663 commit f032916
Show file tree
Hide file tree
Showing 7 changed files with 57 additions and 51 deletions.
42 changes: 21 additions & 21 deletions hapi-fhir-jpaserver-model/src/main/java/ca/uhn/fhir/jpa/model/concurrency/PointcutLatch.java
View file
Expand Up @@ -44,11 +44,11 @@ public class PointcutLatch implements IAnonymousInterceptor, IPointcutLatch {

private final String name;

private CountDownLatch myCountdownLatch;
private AtomicReference<List<String>> myFailures;
private AtomicReference<List<HookParams>> myCalledWith;
private final AtomicReference<CountDownLatch> myCountdownLatch = new AtomicReference<>();
private final AtomicReference<List<String>> myFailures = new AtomicReference<>();
private final AtomicReference<List<HookParams>> myCalledWith = new AtomicReference<>();
private final Pointcut myPointcut;
private int myInitialCount;
private Pointcut myPointcut;

public PointcutLatch(Pointcut thePointcut) {
this.name = thePointcut.name();
Expand All @@ -57,26 +57,27 @@ public PointcutLatch(Pointcut thePointcut) {

public PointcutLatch(String theName) {
this.name = theName;
myPointcut = null;
}

@Override
public void setExpectedCount(int count) {
if (myCountdownLatch != null) {
if (myCountdownLatch.get() != null) {
throw new PointcutLatchException("setExpectedCount() called before previous awaitExpected() completed.");
}
createLatch(count);
ourLog.info("Expecting {} calls to {} latch", count, name);
}

private void createLatch(int count) {
myFailures = new AtomicReference<>(new ArrayList<>());
myCalledWith = new AtomicReference<>(new ArrayList<>());
myCountdownLatch = new CountDownLatch(count);
myFailures.set(new ArrayList<>());
myCalledWith.set(new ArrayList<>());
myCountdownLatch.set(new CountDownLatch(count));
myInitialCount = count;
}

private void addFailure(String failure) {
if (myFailures != null) {
if (myFailures.get() != null) {
myFailures.get().add(failure);
} else {
throw new PointcutLatchException("trying to set failure on latch that hasn't been created: " + failure);
Expand All @@ -95,9 +96,10 @@ public List<HookParams> awaitExpected() throws InterruptedException {
public List<HookParams> awaitExpectedWithTimeout(int timeoutSecond) throws InterruptedException {
List<HookParams> retval = myCalledWith.get();
try {
Validate.notNull(myCountdownLatch, getName() + " awaitExpected() called before setExpected() called.");
if (!myCountdownLatch.await(timeoutSecond, TimeUnit.SECONDS)) {
throw new AssertionError(getName() + " timed out waiting " + timeoutSecond + " seconds for latch to countdown from " + myInitialCount + " to 0. Is " + myCountdownLatch.getCount() + ".");
CountDownLatch latch = myCountdownLatch.get();
Validate.notNull(latch, getName() + " awaitExpected() called before setExpected() called.");
if (!latch.await(timeoutSecond, TimeUnit.SECONDS)) {
throw new AssertionError(getName() + " timed out waiting " + timeoutSecond + " seconds for latch to countdown from " + myInitialCount + " to 0. Is " + latch.getCount() + ".");
}

List<String> failures = myFailures.get();
Expand All @@ -121,11 +123,11 @@ public List<HookParams> awaitExpectedWithTimeout(int timeoutSecond) throws Inter

@Override
public void clear() {
myCountdownLatch = null;
myCountdownLatch.set(null);
}

private String myCalledWithString() {
if (myCalledWith == null) {
if (myCalledWith.get() == null) {
return "[]";
}
List<HookParams> calledWith = myCalledWith.get();
Expand All @@ -140,21 +142,19 @@ private String myCalledWithString() {

@Override
public void invoke(Pointcut thePointcut, HookParams theArgs) {
if (myCountdownLatch == null) {
CountDownLatch latch = myCountdownLatch.get();
if (latch == null) {
throw new PointcutLatchException("invoke() called outside of setExpectedCount() .. awaitExpected(). Probably got more invocations than expected or clear() was called before invoke() arrived.", theArgs);
} else if (myCountdownLatch.getCount() <= 0) {
} else if (latch.getCount() <= 0) {
addFailure("invoke() called when countdown was zero.");
}

if (myCalledWith.get() != null) {
myCalledWith.get().add(theArgs);
}
ourLog.info("Called {} {} with {}", name, myCountdownLatch, hookParamsToString(theArgs));
ourLog.info("Called {} {} with {}", name, latch, hookParamsToString(theArgs));

if (myCountdownLatch == null) {
throw new PointcutLatchException("invoke() called outside of setExpectedCount() .. awaitExpected(). Probably got more invocations than expected or clear() was called before invoke() arrived.", theArgs);
}
myCountdownLatch.countDown();
latch.countDown();
}

public void call(Object arg) {
Expand Down
2 changes: 1 addition & 1 deletion ...cription/src/test/java/ca/uhn/fhir/jpa/subscription/module/BaseSubscriptionDstu3Test.java
View file
Expand Up @@ -14,7 +14,7 @@
@ContextConfiguration(classes = {TestSubscriptionDstu3Config.class})
public abstract class BaseSubscriptionDstu3Test extends BaseSubscriptionTest {

private SubscriptionTestHelper mySubscriptionTestHelper = new SubscriptionTestHelper();
private final SubscriptionTestHelper mySubscriptionTestHelper = new SubscriptionTestHelper();

public static void waitForSize(int theTarget, List<?> theList) {
StopWatch sw = new StopWatch();
Expand Down
8 changes: 0 additions & 8 deletions ...-subscription/src/test/java/ca/uhn/fhir/jpa/subscription/module/BaseSubscriptionTest.java
View file
Expand Up @@ -23,9 +23,6 @@ public abstract class BaseSubscriptionTest {
@Autowired
MockFhirClientSearchParamProvider myMockFhirClientSearchParamProvider;

@Autowired
SubscriptionLoader mySubscriptionLoader;

@Autowired
protected
IInterceptorService myInterceptorRegistry;
Expand All @@ -39,9 +36,4 @@ public void initSearchParamRegistry(IBundleProvider theBundleProvider) {
myMockFhirClientSearchParamProvider.setBundleProvider(theBundleProvider);
mySearchParamRegistry.forceRefresh();
}

public void initSubscriptionLoader(IBundleProvider theBundleProvider) {
myMockFhirClientSubscriptionProvider.setBundleProvider(theBundleProvider);
mySubscriptionLoader.doSyncSubscriptionsForUnitTest();
}
}
2 changes: 1 addition & 1 deletion ...ubscription/src/test/java/ca/uhn/fhir/jpa/subscription/module/SubscriptionTestHelper.java
View file
Expand Up @@ -7,7 +7,7 @@

public class SubscriptionTestHelper {

protected static AtomicLong idCounter = new AtomicLong();
protected static final AtomicLong idCounter = new AtomicLong();


public Subscription makeActiveSubscription(String theCriteria, String thePayload, String theEndpoint) {
Expand Down
32 changes: 23 additions & 9 deletions ...hir/jpa/subscription/module/standalone/BaseBlockingQueueSubscribableChannelDstu3Test.java
View file
Expand Up @@ -9,16 +9,21 @@
import ca.uhn.fhir.jpa.subscription.module.BaseSubscriptionDstu3Test;
import ca.uhn.fhir.jpa.subscription.module.ResourceModifiedMessage;
import ca.uhn.fhir.jpa.subscription.module.cache.SubscriptionChannelFactory;
import ca.uhn.fhir.jpa.subscription.module.cache.SubscriptionLoader;
import ca.uhn.fhir.jpa.subscription.module.cache.SubscriptionRegistry;
import ca.uhn.fhir.jpa.subscription.module.config.MockFhirClientSearchParamProvider;
import ca.uhn.fhir.jpa.subscription.module.config.MockFhirClientSubscriptionProvider;
import ca.uhn.fhir.jpa.subscription.module.subscriber.ResourceModifiedJsonMessage;
import ca.uhn.fhir.jpa.subscription.module.subscriber.SubscriptionMatchingSubscriberTest;
import ca.uhn.fhir.rest.annotation.Create;
import ca.uhn.fhir.rest.annotation.ResourceParam;
import ca.uhn.fhir.rest.annotation.Update;
import ca.uhn.fhir.rest.api.Constants;
import ca.uhn.fhir.rest.api.MethodOutcome;
import ca.uhn.fhir.rest.api.server.IBundleProvider;
import ca.uhn.fhir.rest.server.IResourceProvider;
import ca.uhn.fhir.rest.server.RestfulServer;
import ca.uhn.fhir.rest.server.SimpleBundleProvider;
import ca.uhn.fhir.util.PortUtil;
import com.google.common.collect.Lists;
import org.eclipse.jetty.server.Server;
Expand Down Expand Up @@ -54,31 +59,32 @@ public abstract class BaseBlockingQueueSubscribableChannelDstu3Test extends Base
IInterceptorService myInterceptorRegistry;
@Autowired
protected SubscriptionRegistry mySubscriptionRegistry;

@Autowired
private MockFhirClientSubscriptionProvider myMockFhirClientSubscriptionProvider;
@Autowired
private SubscriptionLoader mySubscriptionLoader;

protected String myCode = "1000000050";

private static int ourListenerPort;
private static RestfulServer ourListenerRestServer;
private static Server ourListenerServer;
protected static String ourListenerServerBase;
protected static List<Observation> ourCreatedObservations = Collections.synchronizedList(Lists.newArrayList());
protected static List<Observation> ourUpdatedObservations = Collections.synchronizedList(Lists.newArrayList());
protected static List<String> ourContentTypes = Collections.synchronizedList(new ArrayList<>());
protected static final List<Observation> ourCreatedObservations = Collections.synchronizedList(Lists.newArrayList());
protected static final List<Observation> ourUpdatedObservations = Collections.synchronizedList(Lists.newArrayList());
protected static final List<String> ourContentTypes = Collections.synchronizedList(new ArrayList<>());
private static SubscribableChannel ourSubscribableChannel;
protected PointcutLatch mySubscriptionMatchingPost = new PointcutLatch(Pointcut.SUBSCRIPTION_AFTER_PERSISTED_RESOURCE_CHECKED);
protected PointcutLatch mySubscriptionActivatedPost = new PointcutLatch(Pointcut.SUBSCRIPTION_AFTER_ACTIVE_SUBSCRIPTION_REGISTERED);
protected final PointcutLatch mySubscriptionMatchingPost = new PointcutLatch(Pointcut.SUBSCRIPTION_AFTER_PERSISTED_RESOURCE_CHECKED);
protected final PointcutLatch mySubscriptionActivatedPost = new PointcutLatch(Pointcut.SUBSCRIPTION_AFTER_ACTIVE_SUBSCRIPTION_REGISTERED);

@Before
public void beforeReset() {
ourCreatedObservations.clear();
ourUpdatedObservations.clear();
ourContentTypes.clear();
mySubscriptionRegistry.unregisterAllSubscriptions();
if (ourSubscribableChannel == null) {
ourSubscribableChannel = mySubscriptionChannelFactory.newDeliveryChannel("test", Subscription.SubscriptionChannelType.RESTHOOK.toCode().toLowerCase());
ourSubscribableChannel.subscribe(myStandaloneSubscriptionMessageHandler);
}
myInterceptorRegistry.registerAnonymousInterceptor(Pointcut.SUBSCRIPTION_AFTER_PERSISTED_RESOURCE_CHECKED, mySubscriptionMatchingPost);
myInterceptorRegistry.registerAnonymousInterceptor(Pointcut.SUBSCRIPTION_AFTER_ACTIVE_SUBSCRIPTION_REGISTERED, mySubscriptionActivatedPost);
}
Expand All @@ -100,6 +106,11 @@ public <T extends IBaseResource> T sendResource(T theResource) throws Interrupte
return theResource;
}

protected void initSubscriptionLoader(List<Subscription> subscriptions, String uuid) throws InterruptedException {
myMockFhirClientSubscriptionProvider.setBundleProvider(new SimpleBundleProvider(new ArrayList<>(subscriptions), uuid));
mySubscriptionLoader.doSyncSubscriptionsForUnitTest();
}

protected Subscription sendSubscription(String theCriteria, String thePayload, String theEndpoint) throws InterruptedException {
Subscription subscription = makeActiveSubscription(theCriteria, thePayload, theEndpoint);
mySubscriptionActivatedPost.setExpectedCount(1);
Expand Down Expand Up @@ -144,6 +155,9 @@ public static void startListenerServer() throws Exception {

ourListenerServer.setHandler(proxyHandler);
ourListenerServer.start();
FhirContext context = ourListenerRestServer.getFhirContext();
//Preload structure definitions so the load doesn't happen during the test (first load can be a little slow)
context.getValidationSupport().fetchAllStructureDefinitions(context);
}

@AfterClass
Expand All @@ -153,7 +167,7 @@ public static void stopListenerServer() throws Exception {

public static class ObservationListener implements IResourceProvider, IPointcutLatch {

private PointcutLatch updateLatch = new PointcutLatch("Observation Update");
private final PointcutLatch updateLatch = new PointcutLatch("Observation Update");

@Create
public MethodOutcome create(@ResourceParam Observation theObservation, HttpServletRequest theRequest) {
Expand Down
13 changes: 7 additions & 6 deletions ...java/ca/uhn/fhir/jpa/subscription/module/standalone/SubscriptionLoaderFhirClientTest.java
View file
@@ -1,8 +1,6 @@
package ca.uhn.fhir.jpa.subscription.module.standalone;

import ca.uhn.fhir.rest.api.Constants;
import ca.uhn.fhir.rest.api.server.IBundleProvider;
import ca.uhn.fhir.rest.server.SimpleBundleProvider;
import org.hl7.fhir.dstu3.model.Subscription;
import org.junit.Test;

Expand All @@ -12,6 +10,7 @@
import static org.junit.Assert.assertEquals;

public class SubscriptionLoaderFhirClientTest extends BaseBlockingQueueSubscribableChannelDstu3Test {

@Test
public void testSubscriptionLoaderFhirClient() throws InterruptedException {
String payload = "application/fhir+json";
Expand All @@ -23,10 +22,13 @@ public void testSubscriptionLoaderFhirClient() throws InterruptedException {
subs.add(makeActiveSubscription(criteria1, payload, ourListenerServerBase));
subs.add(makeActiveSubscription(criteria2, payload, ourListenerServerBase));

IBundleProvider bundle = new SimpleBundleProvider(new ArrayList<>(subs), "uuid");
initSubscriptionLoader(bundle);
mySubscriptionActivatedPost.setExpectedCount(2);
initSubscriptionLoader(subs, "uuid");
mySubscriptionActivatedPost.awaitExpected();

ourObservationListener.setExpectedCount(1);
sendObservation(myCode, "SNOMED-CT");
ourObservationListener.awaitExpected();

waitForSize(0, ourCreatedObservations);
waitForSize(1, ourUpdatedObservations);
Expand All @@ -44,8 +46,7 @@ public void testSubscriptionLoaderFhirClientSubscriptionNotActive() throws Inter
subs.add(makeActiveSubscription(criteria1, payload, ourListenerServerBase).setStatus(Subscription.SubscriptionStatus.REQUESTED));
subs.add(makeActiveSubscription(criteria2, payload, ourListenerServerBase).setStatus(Subscription.SubscriptionStatus.REQUESTED));

IBundleProvider bundle = new SimpleBundleProvider(new ArrayList<>(subs), "uuid");
initSubscriptionLoader(bundle);
initSubscriptionLoader(subs, "uuid");

sendObservation(myCode, "SNOMED-CT");

Expand Down
9 changes: 4 additions & 5 deletions .../src/test/java/ca/uhn/fhir/jpa/subscription/module/standalone/SubscriptionLoaderTest.java
View file
Expand Up @@ -19,8 +19,6 @@ public class SubscriptionLoaderTest extends BaseBlockingQueueSubscribableChannel
private static final int MOCK_FHIR_CLIENT_FAILURES = 5;
@Autowired
private MockFhirClientSubscriptionProvider myMockFhirClientSubscriptionProvider;
@Autowired
private SubscriptionLoader mySubscriptionLoader;

@Before
public void setFailCount() {
Expand All @@ -33,7 +31,7 @@ public void restoreFailCount() {
}

@Test
public void testSubscriptionLoaderFhirClientDown() {
public void testSubscriptionLoaderFhirClientDown() throws Exception {
String payload = "application/fhir+json";

String criteria1 = "Observation?code=SNOMED-CT|" + myCode + "&_format=xml";
Expand All @@ -43,8 +41,9 @@ public void testSubscriptionLoaderFhirClientDown() {
subs.add(makeActiveSubscription(criteria1, payload, ourListenerServerBase));
subs.add(makeActiveSubscription(criteria2, payload, ourListenerServerBase));

IBundleProvider bundle = new SimpleBundleProvider(new ArrayList<>(subs), "uuid");
initSubscriptionLoader(bundle);
mySubscriptionActivatedPost.setExpectedCount(2);
initSubscriptionLoader(subs, "uuid");
mySubscriptionActivatedPost.awaitExpected();
assertEquals(0, myMockFhirClientSubscriptionProvider.getFailCount());
}
}

0 comments on commit f032916

Please sign in to comment.