diff --git a/src/jmh/java/performance/DataLoaderPerformance.java b/src/jmh/java/performance/DataLoaderPerformance.java index d816367716..20e144abd8 100644 --- a/src/jmh/java/performance/DataLoaderPerformance.java +++ b/src/jmh/java/performance/DataLoaderPerformance.java @@ -4,7 +4,6 @@ import graphql.ExecutionInput; import graphql.ExecutionResult; import graphql.GraphQL; -import graphql.execution.instrumentation.dataloader.DataLoaderDispatchingContextKeys; import graphql.schema.DataFetcher; import graphql.schema.GraphQLSchema; import graphql.schema.idl.RuntimeWiring; @@ -15,28 +14,16 @@ import org.dataloader.DataLoader; import org.dataloader.DataLoaderFactory; import org.dataloader.DataLoaderRegistry; -import org.openjdk.jmh.annotations.Benchmark; -import org.openjdk.jmh.annotations.BenchmarkMode; -import org.openjdk.jmh.annotations.Fork; -import org.openjdk.jmh.annotations.Measurement; -import org.openjdk.jmh.annotations.Mode; -import org.openjdk.jmh.annotations.OutputTimeUnit; import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.Setup; import org.openjdk.jmh.annotations.State; -import org.openjdk.jmh.annotations.Warmup; import org.openjdk.jmh.infra.Blackhole; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; -@State(Scope.Benchmark) -@Warmup(iterations = 2, time = 5) -@Measurement(iterations = 3) -@Fork(2) public class DataLoaderPerformance { static Owner o1 = new Owner("O-1", "Andi", List.of("P-1", "P-2", "P-3")); @@ -573,9 +560,6 @@ public void setup() { } - @Benchmark - @BenchmarkMode(Mode.AverageTime) - @OutputTimeUnit(TimeUnit.MILLISECONDS) public void executeRequestWithDataLoaders(MyState myState, Blackhole blackhole) { DataLoader ownerDL = DataLoaderFactory.newDataLoader(ownerBatchLoader); DataLoader petDL = DataLoaderFactory.newDataLoader(petBatchLoader); @@ -587,14 +571,30 @@ public void executeRequestWithDataLoaders(MyState myState, Blackhole blackhole) .dataLoaderRegistry(registry) // .profileExecution(true) .build(); - executionInput.getGraphQLContext().put(DataLoaderDispatchingContextKeys.ENABLE_DATA_LOADER_CHAINING, true); +// executionInput.getGraphQLContext().put(DataLoaderDispatchingContextKeys.ENABLE_DATA_LOADER_CHAINING, true); ExecutionResult execute = myState.graphQL.execute(executionInput); // ProfilerResult profilerResult = executionInput.getGraphQLContext().get(ProfilerResult.PROFILER_CONTEXT_KEY); -// System.out.println(profilerResult.shortSummaryMap()); +// System.out.println("execute: " + execute); Assert.assertTrue(execute.isDataPresent()); Assert.assertTrue(execute.getErrors().isEmpty()); blackhole.consume(execute); } + public static void main(String[] args) { + DataLoaderPerformance dataLoaderPerformance = new DataLoaderPerformance(); + MyState myState = new MyState(); + myState.setup(); + Blackhole blackhole = new Blackhole("Today's password is swordfish. I understand instantiating Blackholes directly is dangerous."); + for (int i = 0; i < 1; i++) { + dataLoaderPerformance.executeRequestWithDataLoaders(myState, blackhole); + } +// System.out.println(PerLevelDataLoaderDispatchStrategy.fieldFetchedCount); +// System.out.println(PerLevelDataLoaderDispatchStrategy.onCompletionFinishedCount); +// System.out.println(PerLevelDataLoaderDispatchStrategy.isReadyCounter); +// System.out.println(Duration.ofNanos(PerLevelDataLoaderDispatchStrategy.isReadyCounterNS.get()).toMillis()); + + + } + } diff --git a/src/main/java/graphql/execution/DataLoaderDispatchStrategy.java b/src/main/java/graphql/execution/DataLoaderDispatchStrategy.java index b3f837cd5c..8763d650f2 100644 --- a/src/main/java/graphql/execution/DataLoaderDispatchStrategy.java +++ b/src/main/java/graphql/execution/DataLoaderDispatchStrategy.java @@ -57,8 +57,11 @@ default void fieldFetched(ExecutionContext executionContext, } + default void newSubscriptionExecution(AlternativeCallContext alternativeCallContext) { - default void newSubscriptionExecution(FieldValueInfo fieldValueInfo, AlternativeCallContext alternativeCallContext) { + } + + default void subscriptionEventCompletionDone(AlternativeCallContext alternativeCallContext) { } } diff --git a/src/main/java/graphql/execution/ExecutionStrategy.java b/src/main/java/graphql/execution/ExecutionStrategy.java index c1272df44b..a6061a15ce 100644 --- a/src/main/java/graphql/execution/ExecutionStrategy.java +++ b/src/main/java/graphql/execution/ExecutionStrategy.java @@ -207,11 +207,11 @@ protected Object executeObject(ExecutionContext executionContext, ExecutionStrat List fieldNames = parameters.getFields().getKeys(); DeferredExecutionSupport deferredExecutionSupport = createDeferredExecutionSupport(executionContext, parameters); + List fieldsExecutedOnInitialResult = deferredExecutionSupport.getNonDeferredFieldNames(fieldNames); + dataLoaderDispatcherStrategy.executeObject(executionContext, parameters, fieldsExecutedOnInitialResult.size()); Async.CombinedBuilder resolvedFieldFutures = getAsyncFieldValueInfo(executionContext, parameters, deferredExecutionSupport); CompletableFuture> overallResult = new CompletableFuture<>(); - List fieldsExecutedOnInitialResult = deferredExecutionSupport.getNonDeferredFieldNames(fieldNames); - dataLoaderDispatcherStrategy.executeObject(executionContext, parameters, fieldsExecutedOnInitialResult.size()); BiConsumer, Throwable> handleResultsConsumer = buildFieldValueMap(fieldsExecutedOnInitialResult, overallResult, executionContext); resolveObjectCtx.onDispatched(); diff --git a/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java b/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java index d2da978471..50cfcb4bca 100644 --- a/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java +++ b/src/main/java/graphql/execution/SubscriptionExecutionStrategy.java @@ -168,9 +168,11 @@ private CompletableFuture executeSubscriptionEvent(ExecutionCon i13nFieldParameters, executionContext.getInstrumentationState() )); + + executionContext.getDataLoaderDispatcherStrategy().newSubscriptionExecution(newParameters.getDeferredCallContext()); Object fetchedValue = unboxPossibleDataFetcherResult(newExecutionContext, newParameters, eventPayload); FieldValueInfo fieldValueInfo = completeField(newExecutionContext, newParameters, fetchedValue); - executionContext.getDataLoaderDispatcherStrategy().newSubscriptionExecution(fieldValueInfo, newParameters.getDeferredCallContext()); + executionContext.getDataLoaderDispatcherStrategy().subscriptionEventCompletionDone(newParameters.getDeferredCallContext()); CompletableFuture overallResult = fieldValueInfo .getFieldValueFuture() .thenApply(val -> new ExecutionResultImpl(val, newParameters.getDeferredCallContext().getErrors())) diff --git a/src/main/java/graphql/execution/instrumentation/dataloader/LevelMap.java b/src/main/java/graphql/execution/instrumentation/dataloader/LevelMap.java index ddf46f643b..45fdca37b9 100644 --- a/src/main/java/graphql/execution/instrumentation/dataloader/LevelMap.java +++ b/src/main/java/graphql/execution/instrumentation/dataloader/LevelMap.java @@ -57,7 +57,7 @@ public String toString() { StringBuilder result = new StringBuilder(); result.append("IntMap["); for (int i = 0; i < countsByLevel.length; i++) { - result.append("level=").append(i).append(",count=").append(countsByLevel[i]).append(" "); + result.append("[level=").append(i).append(",count=").append(countsByLevel[i]).append("] "); } result.append("]"); return result.toString(); diff --git a/src/main/java/graphql/execution/instrumentation/dataloader/PerLevelDataLoaderDispatchStrategy.java b/src/main/java/graphql/execution/instrumentation/dataloader/PerLevelDataLoaderDispatchStrategy.java index bd971f656d..943bae4ffd 100644 --- a/src/main/java/graphql/execution/instrumentation/dataloader/PerLevelDataLoaderDispatchStrategy.java +++ b/src/main/java/graphql/execution/instrumentation/dataloader/PerLevelDataLoaderDispatchStrategy.java @@ -22,6 +22,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -36,7 +37,7 @@ public class PerLevelDataLoaderDispatchStrategy implements DataLoaderDispatchStr private final Profiler profiler; - private final Map deferredCallStackMap = new ConcurrentHashMap<>(); + private final Map alternativeCallContextMap = new ConcurrentHashMap<>(); private static class ChainedDLStack { @@ -134,138 +135,74 @@ public void clear() { private static class CallStack { + static class StateForLevel { + private final int happenedCompletionFinishedCount; + private final int happenedExecuteObjectCalls; - /** - * A general overview of teh tracked data: - * There are three aspects tracked per level: - * - number of execute object calls (executeObject) - * - number of fetches - * - number of sub selections finished fetching - *

- * The level for an execute object call is the level of the field in the query: for - * { a {b {c}}} the level of a is 1, b is 2 and c is not an object - *

- * For fetches the level is the level of the field fetched - *

- * For sub selections finished it is the level of the fields inside the sub selection: - * {a1 { b c} a2 } the level of {a1 a2} is 1, the level of {b c} is 2 - *

- *

- * A finished subselection means we can predict the number of execute object calls in the same level as the subselection: - * { a {x} b {y} } - * If a is a list of 3 objects and b is a list of 2 objects we expect 3 + 2 = 5 execute object calls on the level 1 to be happening - *

- * An executed object call again means we can predict the number of fetches in the next level: - * Execute Object a with { a {f1 f2 f3} } means we expect 3 fetches on level 2. - *

- * This means we know a level is ready to be dispatched if: - * - all subselections done in the parent level - * - all execute objects calls in the parent level are done - * - all expected fetched happened in the current level - */ - - private final LevelMap expectedFetchCountPerLevel = new LevelMap(); - private final LevelMap fetchCountPerLevel = new LevelMap(); - - // an object call means a sub selection of a field of type object/interface/union - // the number of fields for sub selections increases the expected fetch count for this level - private final LevelMap expectedExecuteObjectCallsPerLevel = new LevelMap(); - private final LevelMap happenedExecuteObjectCallsPerLevel = new LevelMap(); - - // this means one sub selection has been fully fetched - // and the expected execute objects calls for the next level have been calculated - private final LevelMap happenedOnFieldValueCallsPerLevel = new LevelMap(); - private final Set dispatchedLevels = ConcurrentHashMap.newKeySet(); - - // all levels that are ready to be dispatched - private int highestReadyLevel; - - public ChainedDLStack chainedDLStack = new ChainedDLStack(); + public StateForLevel() { + this.happenedCompletionFinishedCount = 0; + this.happenedExecuteObjectCalls = 0; + } - private final List deferredFragmentRootFieldsFetched = new ArrayList<>(); + public StateForLevel(int happenedCompletionFinishedCount, int happenedExecuteObjectCalls) { + this.happenedCompletionFinishedCount = happenedCompletionFinishedCount; + this.happenedExecuteObjectCalls = happenedExecuteObjectCalls; + } - public CallStack() { - // in the first level there is only one sub selection, - // so we only expect one execute object call (which is actually an executionStrategy call) - expectedExecuteObjectCallsPerLevel.set(0, 1); - } + public StateForLevel(StateForLevel other) { + this.happenedCompletionFinishedCount = other.happenedCompletionFinishedCount; + this.happenedExecuteObjectCalls = other.happenedExecuteObjectCalls; + } + public StateForLevel copy() { + return new StateForLevel(this); + } - void increaseExpectedFetchCount(int level, int count) { - expectedFetchCountPerLevel.increment(level, count); - } + public StateForLevel increaseHappenedCompletionFinishedCount() { + return new StateForLevel(happenedCompletionFinishedCount + 1, happenedExecuteObjectCalls); + } - void clearExpectedFetchCount() { - expectedFetchCountPerLevel.clear(); - } + public StateForLevel increaseHappenedExecuteObjectCalls() { + return new StateForLevel(happenedCompletionFinishedCount, happenedExecuteObjectCalls + 1); + } - void increaseFetchCount(int level) { - fetchCountPerLevel.increment(level, 1); } + private volatile int expectedFirstLevelFetchCount; + private final AtomicInteger happenedFirstLevelFetchCount = new AtomicInteger(); - void clearFetchCount() { - fetchCountPerLevel.clear(); - } - void increaseExpectedExecuteObjectCalls(int level, int count) { - expectedExecuteObjectCallsPerLevel.increment(level, count); - } + private final Map> stateForLevelMap = new ConcurrentHashMap<>(); - void clearExpectedObjectCalls() { - expectedExecuteObjectCallsPerLevel.clear(); - } + private final Set dispatchedLevels = ConcurrentHashMap.newKeySet(); - void increaseHappenedExecuteObjectCalls(int level) { - happenedExecuteObjectCallsPerLevel.increment(level, 1); - } + public ChainedDLStack chainedDLStack = new ChainedDLStack(); - void clearHappenedExecuteObjectCalls() { - happenedExecuteObjectCallsPerLevel.clear(); - } + private final AtomicInteger deferredFragmentRootFieldsCompleted = new AtomicInteger(); - void increaseHappenedOnFieldValueCalls(int level) { - happenedOnFieldValueCallsPerLevel.increment(level, 1); + public CallStack() { } - void clearHappenedOnFieldValueCalls() { - happenedOnFieldValueCallsPerLevel.clear(); - } - boolean allExecuteObjectCallsHappened(int level) { - return happenedExecuteObjectCallsPerLevel.get(level) == expectedExecuteObjectCallsPerLevel.get(level); + public StateForLevel get(int level) { + AtomicReference dataPerLevelAtomicReference = stateForLevelMap.computeIfAbsent(level, __ -> new AtomicReference<>(new StateForLevel())); + return Assert.assertNotNull(dataPerLevelAtomicReference.get()); } - boolean allSubSelectionsFetchingHappened(int subSelectionLevel) { - return happenedOnFieldValueCallsPerLevel.get(subSelectionLevel) == expectedExecuteObjectCallsPerLevel.get(subSelectionLevel - 1); + public boolean tryUpdateLevel(int level, StateForLevel oldData, StateForLevel newData) { + AtomicReference dataPerLevelAtomicReference = Assert.assertNotNull(stateForLevelMap.get(level)); + return dataPerLevelAtomicReference.compareAndSet(oldData, newData); } - boolean allFetchesHappened(int level) { - return fetchCountPerLevel.get(level) == expectedFetchCountPerLevel.get(level); - } - void clearDispatchLevels() { + public void clear() { dispatchedLevels.clear(); - } - - @Override - public String toString() { - return "CallStack{" + - "expectedFetchCountPerLevel=" + expectedFetchCountPerLevel + - ", fetchCountPerLevel=" + fetchCountPerLevel + - ", expectedExecuteObjectCallsPerLevel=" + expectedExecuteObjectCallsPerLevel + - ", happenedExecuteObjectCallsPerLevel=" + happenedExecuteObjectCallsPerLevel + - ", happenedOnFieldValueCallsPerLevel=" + happenedOnFieldValueCallsPerLevel + - ", dispatchedLevels" + dispatchedLevels + - '}'; - } - - - public void setDispatchedLevel(int level) { - if (!dispatchedLevels.add(level)) { - Assert.assertShouldNeverHappen("level " + level + " already dispatched"); - } + stateForLevelMap.clear(); + expectedFirstLevelFetchCount = 0; + happenedFirstLevelFetchCount.set(0); + deferredFragmentRootFieldsCompleted.set(0); + chainedDLStack.clear(); } } @@ -283,176 +220,90 @@ public PerLevelDataLoaderDispatchStrategy(ExecutionContext executionContext) { @Override public void executionStrategy(ExecutionContext executionContext, ExecutionStrategyParameters parameters, int fieldCount) { Assert.assertTrue(parameters.getExecutionStepInfo().getPath().isRootPath()); - increaseHappenedExecuteObjectAndIncreaseExpectedFetchCount(0, fieldCount, initialCallStack); + // no concurrency access happening + CallStack.StateForLevel currentState = initialCallStack.get(0); + initialCallStack.tryUpdateLevel(0, currentState, new CallStack.StateForLevel(0, 1)); + initialCallStack.expectedFirstLevelFetchCount = fieldCount; } @Override public void executionSerialStrategy(ExecutionContext executionContext, ExecutionStrategyParameters parameters) { CallStack callStack = getCallStack(parameters); - resetCallStack(callStack); - increaseHappenedExecuteObjectAndIncreaseExpectedFetchCount(0, 1, callStack); + callStack.clear(); + CallStack.StateForLevel currentState = initialCallStack.get(0); + initialCallStack.tryUpdateLevel(0, currentState, new CallStack.StateForLevel(0, 1)); + // field count is always 1 for serial execution + initialCallStack.expectedFirstLevelFetchCount = 1; } @Override public void executionStrategyOnFieldValuesInfo(List fieldValueInfoList, ExecutionStrategyParameters parameters) { CallStack callStack = getCallStack(parameters); - // the root fields are the root sub selection on level 1 - onFieldValuesInfoDispatchIfNeeded(fieldValueInfoList, 1, callStack); + onCompletionFinished(0, callStack); + } @Override public void executionStrategyOnFieldValuesException(Throwable t, ExecutionStrategyParameters parameters) { CallStack callStack = getCallStack(parameters); - synchronized (callStack) { - callStack.increaseHappenedOnFieldValueCalls(1); - } - } - - private CallStack getCallStack(ExecutionStrategyParameters parameters) { - return getCallStack(parameters.getDeferredCallContext()); + onCompletionFinished(0, callStack); } - private CallStack getCallStack(@Nullable AlternativeCallContext alternativeCallContext) { - if (alternativeCallContext == null) { - return this.initialCallStack; - } else { - return deferredCallStackMap.computeIfAbsent(alternativeCallContext, k -> { - CallStack callStack = new CallStack(); - int startLevel = alternativeCallContext.getStartLevel(); - int fields = alternativeCallContext.getFields(); - // we make sure that startLevel-1 is considered done - callStack.expectedExecuteObjectCallsPerLevel.set(0, 0); // set to 1 in the constructor of CallStack - callStack.expectedExecuteObjectCallsPerLevel.set(startLevel - 1, 1); - callStack.happenedExecuteObjectCallsPerLevel.set(startLevel - 1, 1); - callStack.highestReadyLevel = startLevel - 1; - callStack.increaseExpectedFetchCount(startLevel, fields); - return callStack; - }); - } - } @Override public void executeObject(ExecutionContext executionContext, ExecutionStrategyParameters parameters, int fieldCount) { CallStack callStack = getCallStack(parameters); int curLevel = parameters.getPath().getLevel(); - increaseHappenedExecuteObjectAndIncreaseExpectedFetchCount(curLevel, fieldCount, callStack); - } - - @Override - public void executeObjectOnFieldValuesInfo - (List fieldValueInfoList, ExecutionStrategyParameters parameters) { - // the level of the sub selection that is fully fetched is one level more than parameters level - int curLevel = parameters.getPath().getLevel() + 1; - CallStack callStack = getCallStack(parameters); - onFieldValuesInfoDispatchIfNeeded(fieldValueInfoList, curLevel, callStack); - } - - - @Override - public void newSubscriptionExecution(FieldValueInfo fieldValueInfo, AlternativeCallContext alternativeCallContext) { - CallStack callStack = getCallStack(alternativeCallContext); - callStack.increaseFetchCount(1); - callStack.deferredFragmentRootFieldsFetched.add(fieldValueInfo); - onFieldValuesInfoDispatchIfNeeded(callStack.deferredFragmentRootFieldsFetched, 1, callStack); + while (true) { + CallStack.StateForLevel currentState = callStack.get(curLevel); + if (callStack.tryUpdateLevel(curLevel, currentState, currentState.increaseHappenedExecuteObjectCalls())) { + return; + } + } } @Override - public void deferredOnFieldValue(String resultKey, FieldValueInfo fieldValueInfo, Throwable - throwable, ExecutionStrategyParameters parameters) { + public void executeObjectOnFieldValuesInfo(List fieldValueInfoList, ExecutionStrategyParameters parameters) { + int curLevel = parameters.getPath().getLevel(); CallStack callStack = getCallStack(parameters); - boolean ready; - synchronized (callStack) { - callStack.deferredFragmentRootFieldsFetched.add(fieldValueInfo); - Assert.assertNotNull(parameters.getDeferredCallContext()); - ready = callStack.deferredFragmentRootFieldsFetched.size() == parameters.getDeferredCallContext().getFields(); - } - if (ready) { - int curLevel = parameters.getPath().getLevel(); - onFieldValuesInfoDispatchIfNeeded(callStack.deferredFragmentRootFieldsFetched, curLevel, callStack); - } + onCompletionFinished(curLevel, callStack); } @Override public void executeObjectOnFieldValuesException(Throwable t, ExecutionStrategyParameters parameters) { CallStack callStack = getCallStack(parameters); - // the level of the sub selection that is errored is one level more than parameters level - int curLevel = parameters.getPath().getLevel() + 1; - synchronized (callStack) { - callStack.increaseHappenedOnFieldValueCalls(curLevel); - } + int curLevel = parameters.getPath().getLevel(); + onCompletionFinished(curLevel, callStack); } - private void increaseHappenedExecuteObjectAndIncreaseExpectedFetchCount(int curLevel, - int fieldCount, - CallStack callStack) { - synchronized (callStack) { - callStack.increaseHappenedExecuteObjectCalls(curLevel); - callStack.increaseExpectedFetchCount(curLevel + 1, fieldCount); - } - } - - private void resetCallStack(CallStack callStack) { - synchronized (callStack) { - callStack.clearDispatchLevels(); - callStack.clearExpectedObjectCalls(); - callStack.clearExpectedFetchCount(); - callStack.clearFetchCount(); - callStack.clearHappenedExecuteObjectCalls(); - callStack.clearHappenedOnFieldValueCalls(); - callStack.expectedExecuteObjectCallsPerLevel.set(0, 1); - callStack.highestReadyLevel = 0; - callStack.chainedDLStack.clear(); - } - } - - private void onFieldValuesInfoDispatchIfNeeded(List fieldValueInfoList, - int subSelectionLevel, - CallStack callStack) { - Integer dispatchLevel; - synchronized (callStack) { - dispatchLevel = handleSubSelectionFetched(fieldValueInfoList, subSelectionLevel, callStack); - } - // the handle on field values check for the next level if it is ready - if (dispatchLevel != null) { - dispatch(dispatchLevel, callStack); + private void onCompletionFinished(int level, CallStack callStack) { + while (true) { + CallStack.StateForLevel currentState = callStack.get(level); + if (callStack.tryUpdateLevel(level, currentState, currentState.increaseHappenedCompletionFinishedCount())) { + break; + } } - } - - // -// thread safety: called with callStack.lock -// - private @Nullable Integer handleSubSelectionFetched(List fieldValueInfos, int subSelectionLevel, CallStack - callStack) { - callStack.increaseHappenedOnFieldValueCalls(subSelectionLevel); - int expectedOnObjectCalls = getObjectCountForList(fieldValueInfos); - // we expect on the level of the current sub selection #expectedOnObjectCalls execute object calls - callStack.increaseExpectedExecuteObjectCalls(subSelectionLevel, expectedOnObjectCalls); - // maybe the object calls happened already (because the DataFetcher return directly values synchronously) - // therefore we check the next levels if they are ready - // this means we could skip some level because the higher level is also already ready, - // which means there is nothing to dispatch on these levels: if x and x+1 is ready, it means there are no - // data loaders used on x - // - // if data loader chaining is disabled (the old algo) the level we dispatch is not really relevant as - // we dispatch the whole registry anyway - - return getHighestReadyLevel(subSelectionLevel + 1, callStack); - } - /** - * the amount of (non nullable) objects that will require an execute object call - */ - private int getObjectCountForList(List fieldValueInfos) { - int result = 0; - for (FieldValueInfo fieldValueInfo : fieldValueInfos) { - if (fieldValueInfo.getCompleteValueType() == FieldValueInfo.CompleteValueType.OBJECT) { - result += 1; - } else if (fieldValueInfo.getCompleteValueType() == FieldValueInfo.CompleteValueType.LIST) { - result += getObjectCountForList(fieldValueInfo.getFieldValueInfos()); + // due to synchronous DataFetcher the completion calls on higher levels + // can happen before the completion calls on lower level + // this means sometimes a lower level completion means multiple levels are ready + // hence this loop here until a level is not ready or already dispatched + int currentLevel = level + 2; + while (true) { + boolean levelReady; + if (callStack.dispatchedLevels.contains(currentLevel)) { + break; } + levelReady = markLevelAsDispatchedIfReady(currentLevel, callStack); + if (levelReady) { + dispatch(currentLevel, callStack); + } else { + break; + } + currentLevel++; } - return result; + } @@ -464,78 +315,111 @@ public void fieldFetched(ExecutionContext executionContext, Supplier dataFetchingEnvironment) { CallStack callStack = getCallStack(executionStrategyParameters); int level = executionStrategyParameters.getPath().getLevel(); - boolean dispatchNeeded; - synchronized (callStack) { - callStack.increaseFetchCount(level); - dispatchNeeded = dispatchIfNeeded(level, callStack); - } - if (dispatchNeeded) { - dispatch(level, callStack); + AlternativeCallContext deferredCallContext = executionStrategyParameters.getDeferredCallContext(); + if (level == 1 || (deferredCallContext != null && level == deferredCallContext.getStartLevel())) { + int happenedFirstLevelFetchCount = callStack.happenedFirstLevelFetchCount.incrementAndGet(); + if (happenedFirstLevelFetchCount == callStack.expectedFirstLevelFetchCount) { + callStack.dispatchedLevels.add(level); + dispatch(level, callStack); + } } - } - // -// thread safety : called with callStack.lock -// - private boolean dispatchIfNeeded(int level, CallStack callStack) { - boolean ready = checkLevelBeingReady(level, callStack); - if (ready) { - callStack.setDispatchedLevel(level); - return true; - } - return false; + @Override + public void newSubscriptionExecution(AlternativeCallContext alternativeCallContext) { + CallStack callStack = new CallStack(); + alternativeCallContextMap.put(alternativeCallContext, callStack); + } - // -// thread safety: called with callStack.lock -// - private @Nullable Integer getHighestReadyLevel(int startFrom, CallStack callStack) { - int curLevel = callStack.highestReadyLevel; + @Override + public void subscriptionEventCompletionDone(AlternativeCallContext alternativeCallContext) { + CallStack callStack = getCallStack(alternativeCallContext); + // this means the single root field is completed (it was never "fetched" because it is + // the event payload) and we can mark level 1 (root fields) as dispatched and level 0 as completed + callStack.dispatchedLevels.add(1); while (true) { - if (!checkLevelImpl(curLevel + 1, callStack)) { - callStack.highestReadyLevel = curLevel; - return curLevel >= startFrom ? curLevel : null; + CallStack.StateForLevel currentState = callStack.get(0); + if (callStack.tryUpdateLevel(0, currentState, currentState.increaseHappenedExecuteObjectCalls())) { + break; } - curLevel++; } + onCompletionFinished(0, callStack); } - private boolean checkLevelBeingReady(int level, CallStack callStack) { - Assert.assertTrue(level > 0); - if (level <= callStack.highestReadyLevel) { - return true; + @Override + public void deferredOnFieldValue(String resultKey, FieldValueInfo fieldValueInfo, Throwable + throwable, ExecutionStrategyParameters parameters) { + CallStack callStack = getCallStack(parameters); + int deferredFragmentRootFieldsCompleted = callStack.deferredFragmentRootFieldsCompleted.incrementAndGet(); + Assert.assertNotNull(parameters.getDeferredCallContext()); + if (deferredFragmentRootFieldsCompleted == parameters.getDeferredCallContext().getFields()) { + onCompletionFinished(parameters.getDeferredCallContext().getStartLevel() - 1, callStack); + } + + } + + + private CallStack getCallStack(ExecutionStrategyParameters parameters) { + return getCallStack(parameters.getDeferredCallContext()); + } + + private CallStack getCallStack(@Nullable AlternativeCallContext alternativeCallContext) { + if (alternativeCallContext == null) { + return this.initialCallStack; + } else { + return alternativeCallContextMap.computeIfAbsent(alternativeCallContext, k -> { + /* + This is only for handling deferred cases. Subscription cases will also get a new callStack, but + it is explicitly created in `newSubscriptionExecution`. + The reason we are doing this lazily is, because we don't have explicit startDeferred callback. + */ + CallStack callStack = new CallStack(); + // on which level the fields are + int startLevel = k.getStartLevel(); + // how many fields are deferred on this level + int fields = k.getFields(); + if (startLevel > 1) { + // parent level is considered dispatched and all fields completed + callStack.dispatchedLevels.add(startLevel - 1); + CallStack.StateForLevel stateForLevel = callStack.get(startLevel - 2); + CallStack.StateForLevel newStateForLevel = stateForLevel.increaseHappenedExecuteObjectCalls().increaseHappenedCompletionFinishedCount(); + callStack.tryUpdateLevel(startLevel - 2, stateForLevel, newStateForLevel); + } + // the parent will have one completion therefore we set the expectation to 1 + CallStack.StateForLevel stateForLevel = callStack.get(startLevel - 1); + callStack.tryUpdateLevel(startLevel - 1, stateForLevel, stateForLevel.increaseHappenedExecuteObjectCalls()); + + // for the current level we set the fetch expectations + callStack.expectedFirstLevelFetchCount = fields; + return callStack; + }); } + } + - for (int i = callStack.highestReadyLevel + 1; i <= level; i++) { - if (!checkLevelImpl(i, callStack)) { + private boolean markLevelAsDispatchedIfReady(int level, CallStack callStack) { + boolean ready = isLevelReady(level, callStack); + if (ready) { + if (!callStack.dispatchedLevels.add(level)) { + // meaning another thread came before us, so they will take care of dispatching return false; } + return true; } - callStack.highestReadyLevel = level; - return true; + return false; } - private boolean checkLevelImpl(int level, CallStack callStack) { - // a level with zero expectations can't be ready - if (callStack.expectedFetchCountPerLevel.get(level) == 0) { - return false; - } - // first we make sure that the expected fetch count is correct - // by verifying that the parent level all execute object + sub selection were fetched - if (!callStack.allExecuteObjectCallsHappened(level - 1)) { - return false; - } - if (level > 1 && !callStack.allSubSelectionsFetchingHappened(level - 1)) { - return false; - } - // the main check: all fetches must have happened - if (!callStack.allFetchesHappened(level)) { - return false; - } - return true; + private boolean isLevelReady(int level, CallStack callStack) { + Assert.assertTrue(level > 1); + // we expect that parent has been dispatched and that all parents fields are completed + // all parent fields completed means all parent parent on completions finished calls must have happened + int happenedExecuteObjectCalls = callStack.get(level - 2).happenedExecuteObjectCalls; + return callStack.dispatchedLevels.contains(level - 1) && + happenedExecuteObjectCalls > 0 && happenedExecuteObjectCalls == callStack.get(level - 2).happenedCompletionFinishedCount; + } void dispatch(int level, CallStack callStack) { diff --git a/src/test/groovy/graphql/execution/instrumentation/dataloader/LevelMapTest.groovy b/src/test/groovy/graphql/execution/instrumentation/dataloader/LevelMapTest.groovy index 8ff6bece5b..1f0069fb19 100644 --- a/src/test/groovy/graphql/execution/instrumentation/dataloader/LevelMapTest.groovy +++ b/src/test/groovy/graphql/execution/instrumentation/dataloader/LevelMapTest.groovy @@ -1,7 +1,6 @@ package graphql.execution.instrumentation.dataloader import spock.lang.Specification -import graphql.AssertException class LevelMapTest extends Specification { @@ -92,13 +91,13 @@ class LevelMapTest extends Specification { sut.increment(0, 42) then: - sut.toString() == "IntMap[level=0,count=42 ]" + sut.toString() == "IntMap[[level=0,count=42] ]" when: sut.increment(1, 1) then: - sut.toString() == "IntMap[level=0,count=42 level=1,count=1 ]" + sut.toString() == "IntMap[[level=0,count=42] [level=1,count=1] ]" } def "can get outside of its size"() {