This is an automated email from the ASF dual-hosted git repository. vpyatkov pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push: new fa238da3a IGNITE-16863 Implement compound VersisionedValue fa238da3a is described below commit fa238da3a209344b13deddd4152b1209b3a93921 Author: Denis Chudov <moongll...@gmail.com> AuthorDate: Fri May 13 10:04:33 2022 +0300 IGNITE-16863 Implement compound VersisionedValue --- .../ignite/internal/causality/VersionedValue.java | 437 ++++++++++++--------- .../org/apache/ignite/lang/IgniteTriConsumer.java | 57 +++ .../internal/causality/VersionedValueTest.java | 352 ++++++++++++++--- .../runner/app/ItIgniteNodeRestartTest.java | 12 +- .../org/apache/ignite/internal/app/IgniteImpl.java | 10 +- .../internal/sql/engine/SqlQueryProcessor.java | 5 +- .../sql/engine/schema/SqlSchemaManagerImpl.java | 113 +++--- .../internal/sql/engine/StopCalciteModuleTest.java | 21 +- .../sql/engine/exec/MockedStructuresTest.java | 19 +- .../engine/exec/schema/SqlSchemaManagerTest.java | 21 +- .../internal/table/distributed/TableManager.java | 81 ++-- .../ignite/internal/table/TableManagerTest.java | 11 +- 12 files changed, 784 insertions(+), 355 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java b/modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java index 9e7b8d8ed..4109f2a00 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/causality/VersionedValue.java @@ -17,20 +17,28 @@ package org.apache.ignite.internal.causality; +import static java.util.concurrent.CompletableFuture.completedFuture; +import static java.util.concurrent.CompletableFuture.failedFuture; + +import java.util.ArrayList; +import java.util.List; import java.util.Map.Entry; import java.util.concurrent.CancellationException; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentNavigableMap; import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; -import java.util.function.BiConsumer; +import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import org.apache.ignite.lang.IgniteInternalException; import org.apache.ignite.lang.IgniteStringFormatter; -import org.jetbrains.annotations.Nullable; +import org.apache.ignite.lang.IgniteTriConsumer; /** * Parametrized type to store several versions of the value. @@ -45,14 +53,11 @@ public class VersionedValue<T> { /** Default history size. */ private static final int DEFAULT_HISTORY_SIZE = 2; - /** Last applied causality token. */ - private volatile long actualToken = NOT_INITIALIZED; - /** Size of stored history. */ private final int historySize; - /** Closure applied on storage revision update. */ - private final BiConsumer<VersionedValue<T>, Long> storageRevisionUpdateCallback; + /** List of completion listeners, see {@link #whenComplete(IgniteTriConsumer)}. */ + private final List<IgniteTriConsumer<Long, T, Throwable>> completionListeners = new CopyOnWriteArrayList<>(); /** Versioned value storage. */ private final ConcurrentNavigableMap<Long, CompletableFuture<T>> history = new ConcurrentSkipListMap<>(); @@ -63,101 +68,88 @@ public class VersionedValue<T> { */ private final ReadWriteLock trimHistoryLock = new ReentrantReadWriteLock(); - /** Temporary value for {@link #update(long, Function, Function)}. */ - private volatile T tempValue = null; - - /** Whether {@link #update(long, Function, Function)} was called since the last revision update. */ - private volatile boolean isUpdating = false; - - /** Update mutex. */ - private final Object updateMutex = new Object(); - /** Initial future. The future will be completed when {@link VersionedValue} sets a first value. */ private final CompletableFuture<T> initFut = new CompletableFuture<>(); /** The supplier may provide a value which will used as a default. */ private final Supplier<T> defaultValSupplier; + /** Update mutex. */ + private final Object updateMutex = new Object(); + /** Value that can be used as default. */ - private volatile T defaultVal; + private final AtomicReference<T> defaultValRef; + + /** Last applied causality token. */ + private volatile long actualToken = NOT_INITIALIZED; + + /** + * Future that will be completed after all updates over the value in context of current causality token will be performed. + * This {@code updaterFuture} is {@code null} if no updates in context of current causality token have been initiated. + * See {@link #update(long, BiFunction)}. + */ + private volatile CompletableFuture<T> updaterFuture = null; /** * Constructor. * - * @param storageRevisionUpdateCallback Closure applied on storage revision update (see {@link #onStorageRevisionUpdate(long)}). * @param observableRevisionUpdater A closure intended to connect this VersionedValue with a revision updater, that this * VersionedValue should be able to listen to, for receiving storage revision updates. * This closure is called once on a construction of this VersionedValue and accepts a - * {@code Consumer<Long>} that should be called on every update of storage revision as a - * listener. IMPORTANT: Revision update shouldn't happen concurrently with {@link #set(long, T)} - * operations. + * {@code Function<Long, CompletableFuture<?>>} that should be called on every update of + * storage revision as a listener. IMPORTANT: Revision update shouldn't happen + * concurrently with {@link #complete(long, T)} operations. * @param historySize Size of the history of changes to store, including last applied token. - * @param defaultVal Supplier of the default value, that is used on {@link #update(long, Function, Function)} to - * evaluate the default value if the value is not initialized yet. + * @param defaultVal Supplier of the default value, that is used on {@link #update(long, BiFunction)} to + * evaluate the default value if the value is not initialized yet. It is not guaranteed to + * execute only once. */ public VersionedValue( - @Nullable BiConsumer<VersionedValue<T>, Long> storageRevisionUpdateCallback, - Consumer<Consumer<Long>> observableRevisionUpdater, + Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater, int historySize, Supplier<T> defaultVal ) { - this.storageRevisionUpdateCallback = storageRevisionUpdateCallback; - this.historySize = historySize; this.defaultValSupplier = defaultVal; - observableRevisionUpdater.accept(this::onStorageRevisionUpdate); + this.defaultValRef = defaultValSupplier == null ? null : new AtomicReference<>(); + + observableRevisionUpdater.accept(this::completeOnRevision); } /** * Constructor. * - * @param observableRevisionUpdater A closure intended to connect this VersionedValue with a revision updater, that this - * VersionedValue should be able to listen to, for receiving storage revision updates. - * This closure is called once on a construction of this VersionedValue and accepts a - * {@code Consumer<Long>} that should be called on every update of storage revision as a - * listener. IMPORTANT: Revision update shouldn't happen concurrently with {@link #set(long, T)} - * operations. - * @param defaultVal Supplier of the default value, that is used on {@link #update(long, Function, Function)} to - * evaluate the default value if the value is not initialized yet. + * @param observableRevisionUpdater A closure intended to connect this VersionedValue with a revision updater, that this + * VersionedValue should be able to listen to, for receiving storage revision updates. + * This closure is called once on a construction of this VersionedValue and accepts a + * {@code Function<Long, CompletableFuture<?>>} that should be called on every update of + * storage revision as a listener. IMPORTANT: Revision update shouldn't happen + * concurrently with {@link #complete(long, T)} operations. + * @param defaultVal Supplier of the default value, that is used on {@link #update(long, BiFunction)} to + * evaluate the default value if the value is not initialized yet. It is not guaranteed to + * execute only once. */ public VersionedValue( - Consumer<Consumer<Long>> observableRevisionUpdater, + Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater, Supplier<T> defaultVal ) { - this(null, observableRevisionUpdater, DEFAULT_HISTORY_SIZE, defaultVal); + this(observableRevisionUpdater, DEFAULT_HISTORY_SIZE, defaultVal); } /** - * Constructor with default history size that equals 2. See {@link #VersionedValue(BiConsumer, Consumer, int, Supplier)}. + * Constructor with default history size that equals 2. See {@link #VersionedValue(Consumer, int, Supplier)}. * - * @param storageRevisionUpdateCallback Closure applied on storage revision update (see {@link #onStorageRevisionUpdate(long)}. * @param observableRevisionUpdater A closure intended to connect this VersionedValue with a revision updater, that this * VersionedValue should be able to listen to, for receiving storage revision updates. * This closure is called once on a construction of this VersionedValue and accepts a - * {@code Consumer<Long>} that should be called on every update of storage revision as a - * listener. IMPORTANT: Revision update shouldn't happen concurrently with - * {@link #set(long, T)} and {@link #update(long, Function, Function)} operations. + * {@code Function<Long, CompletableFuture<?>>} that should be called on every update of + * storage revision as a listener. IMPORTANT: Revision update shouldn't happen + * concurrently with {@link #complete(long, T)} operations. */ - public VersionedValue( - @Nullable BiConsumer<VersionedValue<T>, Long> storageRevisionUpdateCallback, - Consumer<Consumer<Long>> observableRevisionUpdater - ) { - this(storageRevisionUpdateCallback, observableRevisionUpdater, DEFAULT_HISTORY_SIZE, null); - } - - /** - * Constructor with default history size that equals 2 and no closure. See {@link #VersionedValue(BiConsumer, Consumer, int, Supplier)}. - * - * @param observableRevisionUpdater A closure intended to connect this VersionedValue with a revision updater, that this VersionedValue - * should be able to listen to, for receiving storage revision updates. This closure is called once on - * a construction of this VersionedValue and accepts a {@code Consumer<Long>} that should be called - * on every update of storage revision as a listener. IMPORTANT: Revision update shouldn't happen - * concurrently with {@link #set(long, T)} operations. - */ - public VersionedValue(Consumer<Consumer<Long>> observableRevisionUpdater) { - this(null, observableRevisionUpdater); + public VersionedValue(Consumer<Function<Long, CompletableFuture<?>>> observableRevisionUpdater) { + this(observableRevisionUpdater, DEFAULT_HISTORY_SIZE, null); } /** @@ -190,18 +182,20 @@ public class VersionedValue<T> { * @return The future. */ private CompletableFuture<T> getInternal(long causalityToken) { + long actualToken0 = this.actualToken; + if (history.floorEntry(causalityToken) == null) { - throw new OutdatedTokenException(causalityToken, actualToken, historySize); + throw new OutdatedTokenException(causalityToken, actualToken0, historySize); } - if (causalityToken <= actualToken) { + if (causalityToken <= actualToken0) { return getValueForPreviousToken(causalityToken); } trimHistoryLock.readLock().lock(); try { - if (causalityToken <= actualToken) { + if (causalityToken <= actualToken0) { return getValueForPreviousToken(causalityToken); } @@ -225,31 +219,31 @@ public class VersionedValue<T> { } } - synchronized (updateMutex) { - return getDefault(); - } + return getDefault(); } /** - * Returns a default value. + * Creates (if needed) and returns a default value. * * @return The value. */ private T getDefault() { - // It is thread safe, as it's protected by either updateMutex or exclusiveness of #onStorageRevisionUpdate - // in all usages of getDefault(). - if (defaultValSupplier != null && defaultVal == null) { - defaultVal = defaultValSupplier.get(); + if (defaultValSupplier != null && defaultValRef.get() == null) { + T defaultVal = defaultValSupplier.get(); + + assert defaultVal != null : "Default value can't be null."; + + defaultValRef.compareAndSet(null, defaultVal); } - return defaultVal; + return defaultValRef == null ? null : defaultValRef.get(); } /** * Gets a value for less or equal token than the actual {@link #actualToken}. * * @param causalityToken Causality token. - * @return A completed future that contained a value. + * @return A completed future that contains a value. * @throws OutdatedTokenException If outdated token is passed as an argument. */ private CompletableFuture<T> getValueForPreviousToken(long causalityToken) { @@ -262,6 +256,16 @@ public class VersionedValue<T> { return histEntry.getValue(); } + /** + * Save the version of the value associated with the given causality token. If someone has got a future to await the value associated + * with the given causality token (see {@link #get(long)}, then the future will be completed. + * + * @param causalityToken Causality token. + */ + public void complete(long causalityToken) { + completeOnRevision(causalityToken); + } + /** * Save the version of the value associated with the given causality token. If someone has got a future to await the value associated * with the given causality token (see {@link #get(long)}, then the future will be completed. @@ -269,48 +273,103 @@ public class VersionedValue<T> { * @param causalityToken Causality token. * @param value Current value. */ - public void set(long causalityToken, T value) { + public void complete(long causalityToken, T value) { long actualToken0 = actualToken; if (actualToken0 == NOT_INITIALIZED) { history.put(causalityToken, initFut); } - assert actualToken0 == NOT_INITIALIZED || actualToken0 + 1 == causalityToken : IgniteStringFormatter.format( - "Token must be greater than actual by exactly 1 [token={}, actual={}]", causalityToken, actualToken0); + checkToken(actualToken0, causalityToken); - setValueInternal(causalityToken, value); + completeInternal(causalityToken, value, null); } /** - * Comparisons an exception to the causality token. + * Save the exception associated with the given causality token. If someone has got a future to await the value associated + * with the given causality token (see {@link #get(long)}, then the future will be completed. * * @param causalityToken Causality token. * @param throwable An exception. */ - public void fail(long causalityToken, Throwable throwable) { + public void completeExceptionally(long causalityToken, Throwable throwable) { long actualToken0 = actualToken; if (actualToken0 == NOT_INITIALIZED) { history.put(causalityToken, initFut); } - assert actualToken0 == NOT_INITIALIZED || actualToken0 + 1 == causalityToken : IgniteStringFormatter.format( - "Token must be greater than actual by exactly 1 [token={}, actual={}]", causalityToken, actualToken0); + checkToken(actualToken0, causalityToken); + + completeInternal(causalityToken, null, throwable); + } + + /** + * This internal method assigns either value or exception according to specific token. + * + * @param causalityToken Causality token. + * @param value Value to set. + * @param throwable An exception. + */ + private void completeInternal(long causalityToken, T value, Throwable throwable) { + CompletableFuture<T> res = history.putIfAbsent( + causalityToken, + throwable == null ? completedFuture(value) : failedFuture(throwable) + ); + + if (res == null) { + notifyCompletionListeners(causalityToken, value, throwable); + + return; + } + + assert !res.isDone() : completeInternalConflictErrorMessage(res, causalityToken, value, throwable); + + if (throwable == null) { + res.complete(value); + } else { + res.completeExceptionally(throwable); + } + + notifyCompletionListeners(causalityToken, value, throwable); + } - failInternal(causalityToken, throwable); + /** + * Builds an error message for the case when there is a conflict between history and a value or exception that is going to be + * saved. + * + * @param future Future. + * @param token Token. + * @param value Value. + * @param throwable Throwable. + * @return Error message. + */ + private String completeInternalConflictErrorMessage(CompletableFuture<T> future, long token, T value, Throwable throwable) { + return future.handle( + (prevValue, prevThrowable) -> + IgniteStringFormatter.format( + "Different values associated with the token [token={}, value={}, exception={}, prevValue={}, prevException={}]", + token, + value, + throwable, + prevValue, + prevThrowable + ) + ) + .join(); } /** * Updates the value using the given updater. The updater receives the value on previous token, or default value * (see constructor) if the value isn't initialized, or current intermediate value, if this method has been already * called for the same token; and returns a new value.<br> + * The updater will be called after updaters that had been passed to previous calls of this method complete. * If an exception ({@link CancellationException} or {@link CompletionException}) was thrown when calculating the value for previous - * token, then {@code fail} updater is used to process the exception and calculate a new value.<br> + * token, then updater is used to process the exception and calculate a new value.<br> * This method can be called multiple times for the same token, and doesn't complete the future created for this token. - * The future is supposed to be completed by storage revision update in this case. If this method has been called at least - * once on the given token, the updater will receive a value that was evaluated by updater on previous call, as intermediate - * result.<br> + * The future is supposed to be completed by storage revision update or a call of {@link #complete(long)} in this case. + * If this method has been called at least once on the given token, the updater will receive a value that was evaluated + * by updater on previous call, as intermediate result.<br> * As the order of multiple calls of this method on the same token is unknown, operations done by the updater must be * commutative. For example: * <ul> @@ -322,107 +381,95 @@ public class VersionedValue<T> { * </ul> * Regardless of order in which this method's calls are made, V3 should be the final result. * <br> - * The method should return previous value (previous intermediate value, or a value for previous token, if this method - * is called for first time for given token). + * The method should return a future that will be completed when {@code updater} completes. * * @param causalityToken Causality token. - * @param complete The function is invoked if the previous future completed successfully. - * @param fail The function is invoked if the previous future completed with an exception. - * @return Updated value. + * @param updater The binary function that accepts previous value and exception, if present, and update it to compute + * the new value. + * @return Future for updated value. */ - public T update(long causalityToken, Function<T, T> complete, Function<Throwable, T> fail) { - long actualToken0 = actualToken; - - assert actualToken0 == NOT_INITIALIZED || actualToken0 + 1 == causalityToken : IgniteStringFormatter.format( - "Token must be greater than actual by exactly 1 [token={}, actual={}]", causalityToken, actualToken0); - - try { - synchronized (updateMutex) { - T previousValue; - - if (isUpdating) { - previousValue = tempValue; - } else { - Entry<Long, CompletableFuture<T>> histEntry = history.floorEntry(actualToken0); - - if (histEntry == null) { - previousValue = getDefault(); - } else { - assert histEntry.getValue().isDone() : "Previous value should be ready."; - - previousValue = histEntry.getValue().join(); - } - } + public CompletableFuture<T> update( + long causalityToken, + BiFunction<T, Throwable, CompletableFuture<T>> updater + ) { + long actualToken0 = this.actualToken; - isUpdating = true; + checkToken(actualToken0, causalityToken); - T res = complete.apply(previousValue); + synchronized (updateMutex) { + CompletableFuture<T> updaterFuture = this.updaterFuture; - tempValue = res; + CompletableFuture<T> future = updaterFuture == null ? previousOrDefaultValueFuture(actualToken0) : updaterFuture; - return res; - } - } catch (CancellationException | CompletionException e) { - synchronized (updateMutex) { - isUpdating = true; + CompletableFuture<CompletableFuture<T>> f0 = future + .handle(updater::apply) + .handle((fut, e) -> e == null ? fut : failedFuture(e)); - T res = fail.apply(e); + updaterFuture = f0.thenCompose(Function.identity()); - tempValue = res; + this.updaterFuture = updaterFuture; - return res; - } + return updaterFuture; } } /** - * This internal method assigns value according to specific token without additional checks. + * Add listener for completions of this versioned value on every token. It will be called on every {@link #complete(long)}, + * {@link #complete(long, Object)}, {@link #completeExceptionally(long, Throwable)} and also, if none of mentioned methods was + * called explicitly, on storage revision update, * - * @param causalityToken Causality token. - * @param value Value to set. + * @param action Action to perform. */ - private void setValueInternal(long causalityToken, T value) { - CompletableFuture<T> res = history.putIfAbsent(causalityToken, CompletableFuture.completedFuture(value)); - - if (res == null || res.isCompletedExceptionally()) { - return; - } - - assert !res.isDone() : IgniteStringFormatter.format("Different values associated with the token " - + "[token={}, value={}, prevValue={}]", causalityToken, value, res.join()); + public void whenComplete(IgniteTriConsumer<Long, T, Throwable> action) { + completionListeners.add(action); + } - res.complete(value); + /** + * Removes a completion listener, see {@link #whenComplete(IgniteTriConsumer)}. + * + * @param action Action to remove. + */ + public void removeWhenComplete(IgniteTriConsumer<Long, T, Throwable> action) { + completionListeners.remove(action); } /** - * Fails a future associated with this causality token. + * Notify completion listeners. * - * @param causalityToken Causality token. - * @param throwable An exception. + * @param causalityToken Token. + * @param value Value. + * @param throwable Throwable. */ - private void failInternal(long causalityToken, Throwable throwable) { - CompletableFuture<T> res = history.putIfAbsent(causalityToken, CompletableFuture.failedFuture(throwable)); + private void notifyCompletionListeners(long causalityToken, T value, Throwable throwable) { + Throwable unpackedThrowable = throwable instanceof CompletionException ? throwable.getCause() : throwable; - if (res == null || res.isCompletedExceptionally()) { - return; + List<Exception> exceptions = new ArrayList<>(); + + for (IgniteTriConsumer<Long, T, Throwable> listener : completionListeners) { + try { + listener.accept(causalityToken, value, unpackedThrowable); + } catch (Exception e) { + exceptions.add(e); + } } - assert !res.isDone() : IgniteStringFormatter.format("A value already has associated with the token " - + "[token={}, ex={}, value={}]", causalityToken, throwable, res.join()); + if (!exceptions.isEmpty()) { + IgniteInternalException ex = new IgniteInternalException(); + + exceptions.forEach(ex::addSuppressed); - res.completeExceptionally(throwable); + throw ex; + } } /** - * Should be called on a storage revision update. This also triggers completion of a future created for the given causality token. It - * implies that all possible updates associated with this token have been already applied to the component. - * <br> - * This method should not be called concurrently with {@link #update(long, Function, Function)} and {@link #set(long, Object)} - * methods, as the storage revision update listener is supposed to be called after all other configuration listeners. + * Complete because of explicit token update. This also triggers completion of a future created for the given causality token. + * This future completes after all updaters are complete (see {@link #update(long, BiFunction)}). * - * @param causalityToken Causality token. + * @param causalityToken Token. + * @return Future. */ - private void onStorageRevisionUpdate(long causalityToken) { + private CompletableFuture<?> completeOnRevision(long causalityToken) { long actualToken0 = actualToken; assert causalityToken > actualToken0 : IgniteStringFormatter.format( @@ -432,32 +479,34 @@ public class VersionedValue<T> { history.put(causalityToken, initFut); } - if (isUpdating) { - setValueInternal(causalityToken, tempValue); + synchronized (updateMutex) { + CompletableFuture<T> updaterFuture0 = updaterFuture; - isUpdating = false; - } + CompletableFuture<?> completeUpdatesFuture = updaterFuture0 == null + ? completedFuture(null) + : updaterFuture0.whenComplete((v, t) -> completeInternal(causalityToken, v, t)); - if (storageRevisionUpdateCallback != null) { - storageRevisionUpdateCallback.accept(this, causalityToken); - } + updaterFuture = null; - completeRelatedFuture(causalityToken); + actualToken = causalityToken; - if (history.size() > 1 && causalityToken - history.firstKey() >= historySize) { - trimToSize(causalityToken); - } + return completeUpdatesFuture.thenRun(() -> { + completeRelatedFuture(causalityToken); - Entry<Long, CompletableFuture<T>> entry = history.floorEntry(causalityToken); + if (history.size() > 1 && causalityToken - history.firstKey() >= historySize) { + trimToSize(causalityToken); + } - assert entry != null && entry.getValue().isDone() : IgniteStringFormatter.format( - "Future for the token is not completed [token={}]", causalityToken); + Entry<Long, CompletableFuture<T>> entry = history.floorEntry(causalityToken); - actualToken = causalityToken; + assert entry != null && entry.getValue().isDone() : IgniteStringFormatter.format( + "Future for the token is not completed [token={}]", causalityToken); + }); + } } /** - * Completes a future related with a specific causality token. + * Completes a future related with a specific causality token. It is called only on storage revision update. * * @param causalityToken The token which is becoming an actual. */ @@ -469,18 +518,45 @@ public class VersionedValue<T> { if (!future.isDone()) { Entry<Long, CompletableFuture<T>> entryBefore = history.headMap(causalityToken).lastEntry(); - CompletableFuture<T> previousFuture = - entryBefore == null ? CompletableFuture.completedFuture(getDefault()) : entryBefore.getValue(); + CompletableFuture<T> previousFuture = entryBefore == null ? completedFuture(getDefault()) : entryBefore.getValue(); assert previousFuture.isDone() : IgniteStringFormatter.format("No future for token [token={}]", causalityToken); previousFuture.whenComplete((t, throwable) -> { if (throwable != null) { future.completeExceptionally(throwable); + + notifyCompletionListeners(causalityToken, null, throwable); } else { future.complete(t); + + notifyCompletionListeners(causalityToken, t, null); } }); + } else if (entry.getKey() < causalityToken) { + // Notifying listeners when there were neither updates nor explicit completions. + // This future is previous, it is always done. + future.whenComplete((v, e) -> notifyCompletionListeners(causalityToken, v, e)); + } + } + + /** + * Return a future for previous or default value. + * + * @param actualToken Token. + * @return Future. + */ + private CompletableFuture<T> previousOrDefaultValueFuture(long actualToken) { + Entry<Long, CompletableFuture<T>> histEntry = history.floorEntry(actualToken); + + if (histEntry == null) { + return completedFuture(getDefault()); + } else { + CompletableFuture<T> prevFuture = histEntry.getValue(); + + assert prevFuture.isDone() : "Previous value should be ready."; + + return prevFuture; } } @@ -504,4 +580,15 @@ public class VersionedValue<T> { trimHistoryLock.writeLock().unlock(); } } + + /** + * Check that the given causality token os correct according to the actual token. + * + * @param actualToken Actual token. + * @param causalityToken Causality token. + */ + private static void checkToken(long actualToken, long causalityToken) { + assert actualToken == NOT_INITIALIZED || actualToken + 1 == causalityToken : IgniteStringFormatter.format( + "Token must be greater than actual by exactly 1 [token={}, actual={}]", causalityToken, actualToken); + } } diff --git a/modules/core/src/main/java/org/apache/ignite/lang/IgniteTriConsumer.java b/modules/core/src/main/java/org/apache/ignite/lang/IgniteTriConsumer.java new file mode 100644 index 000000000..8d56b0da6 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/lang/IgniteTriConsumer.java @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.lang; + +import java.util.Objects; + +/** + * Represents a closure that accepts three input parameters and returns no result. + * + * @param <T> The type of the first parameter to the closure. + * @param <U> The type of the second parameter to the closure. + * @param <V> The type of the third parameter to the closure. + */ +@FunctionalInterface +public interface IgniteTriConsumer<T, U, V> { + /** + * Performs the operation on the given arguments. + * + * @param t The first argument. + * @param u The second argument. + * @param v The third argument. + */ + void accept(T t, U u, V v); + + /** + * Returns a composed {@code IgniteTriConsumer} that performs, in sequence, this operation followed by the {@code after} closure. + * If {@code after} throws an exception, it should be handled by the caller of the closure. If performing this closure throws + * an exception, the {@code after} closure will not be performed. + * + * @param after Closure to execute after this {@code IgniteTriConsumer} completes. + * @return Composed {@code IgniteTriConsumer}. + */ + default IgniteTriConsumer<T, U, V> andThen(IgniteTriConsumer<? super T, ? super U, ? super V> after) { + Objects.requireNonNull(after); + + return (t, u, v) -> { + accept(t, u, v); + + after.accept(t, u, v); + }; + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/causality/VersionedValueTest.java b/modules/core/src/test/java/org/apache/ignite/internal/causality/VersionedValueTest.java index 904aa81b4..791e1138c 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/causality/VersionedValueTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/causality/VersionedValueTest.java @@ -17,17 +17,32 @@ package org.apache.ignite.internal.causality; +import static java.util.concurrent.CompletableFuture.completedFuture; +import static java.util.concurrent.CompletableFuture.failedFuture; +import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertThrowsExactly; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiFunction; import java.util.function.Consumer; +import java.util.function.Function; import org.apache.ignite.lang.IgniteInternalException; +import org.apache.ignite.lang.IgniteTriConsumer; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; /** @@ -41,6 +56,11 @@ public class VersionedValueTest { /** The test revision register is used to move the revision forward. */ public static final TestRevisionRegister REGISTER = new TestRevisionRegister(); + @BeforeEach + public void clearRegister() { + REGISTER.clear(); + } + /** * The test gets a value for {@link VersionedValue} before the value is calculated. * @@ -48,26 +68,21 @@ public class VersionedValueTest { */ @Test public void testGetValueBeforeReady() throws OutdatedTokenException { - VersionedValue<Integer> longVersionedValue = new VersionedValue<>( - (integerVersionedValue, token) -> { - integerVersionedValue.set(token, TEST_VALUE); - }, - REGISTER, - 2, - null - ); + VersionedValue<Integer> intVersionedValue = new VersionedValue<>(REGISTER, 2, null); - CompletableFuture<Integer> fut = longVersionedValue.get(0); + CompletableFuture<Integer> fut = intVersionedValue.get(0); assertFalse(fut.isDone()); - REGISTER.moveRevision.accept(0L); + intVersionedValue.complete(0L, TEST_VALUE); + + REGISTER.moveRevision(0L).join(); assertTrue(fut.isDone()); assertEquals(TEST_VALUE, fut.join()); - assertSame(fut.join(), longVersionedValue.get(0).join()); + assertSame(fut.join(), intVersionedValue.get(0).join()); } /** @@ -83,7 +98,7 @@ public class VersionedValueTest { assertFalse(fut.isDone()); - longVersionedValue.set(0, TEST_VALUE); + longVersionedValue.complete(0, TEST_VALUE); assertTrue(fut.isDone()); @@ -102,15 +117,15 @@ public class VersionedValueTest { public void testMissValueUpdateBeforeReady() throws OutdatedTokenException { VersionedValue<Integer> longVersionedValue = new VersionedValue<>(REGISTER); - longVersionedValue.set(0, TEST_VALUE); + longVersionedValue.complete(0, TEST_VALUE); - REGISTER.moveRevision.accept(0L); + REGISTER.moveRevision(0L).join(); CompletableFuture<Integer> fut = longVersionedValue.get(1); assertFalse(fut.isDone()); - REGISTER.moveRevision.accept(1L); + REGISTER.moveRevision(1L).join(); assertTrue(fut.isDone()); @@ -129,10 +144,10 @@ public class VersionedValueTest { public void testMissValueUpdate() throws OutdatedTokenException { VersionedValue<Integer> longVersionedValue = new VersionedValue<>(REGISTER); - longVersionedValue.set(0, TEST_VALUE); + longVersionedValue.complete(0, TEST_VALUE); - REGISTER.moveRevision.accept(0L); - REGISTER.moveRevision.accept(1L); + REGISTER.moveRevision(0L).join(); + REGISTER.moveRevision(1L).join(); CompletableFuture<Integer> fut = longVersionedValue.get(1); @@ -150,14 +165,14 @@ public class VersionedValueTest { public void testObsoleteToken() { VersionedValue<Integer> longVersionedValue = new VersionedValue<>(REGISTER); - longVersionedValue.set(0, TEST_VALUE); + longVersionedValue.complete(0, TEST_VALUE); - REGISTER.moveRevision.accept(0L); + REGISTER.moveRevision(0L).join(); - longVersionedValue.set(1, TEST_VALUE); + longVersionedValue.complete(1, TEST_VALUE); - REGISTER.moveRevision.accept(1L); - REGISTER.moveRevision.accept(2L); + REGISTER.moveRevision(1L).join(); + REGISTER.moveRevision(2L).join(); assertThrowsExactly(OutdatedTokenException.class, () -> longVersionedValue.get(0)); } @@ -167,19 +182,18 @@ public class VersionedValueTest { */ @Test public void testAutocompleteFuture() throws OutdatedTokenException { - VersionedValue<Integer> longVersionedValue = new VersionedValue<>((b, r) -> { - }, REGISTER); + VersionedValue<Integer> longVersionedValue = new VersionedValue<>(REGISTER); - longVersionedValue.set(0, TEST_VALUE); + longVersionedValue.complete(0, TEST_VALUE); - REGISTER.moveRevision.accept(0L); + REGISTER.moveRevision(0L).join(); CompletableFuture<Integer> fut = longVersionedValue.get(1); assertFalse(fut.isDone()); - REGISTER.moveRevision.accept(1L); - REGISTER.moveRevision.accept(2L); + REGISTER.moveRevision(1L).join(); + REGISTER.moveRevision(2L).join(); assertTrue(fut.isDone()); assertTrue(longVersionedValue.get(2).isDone()); @@ -192,12 +206,11 @@ public class VersionedValueTest { */ @Test public void testUpdate() throws Exception { - VersionedValue<Integer> longVersionedValue = new VersionedValue<>((b, r) -> { - }, REGISTER); + VersionedValue<Integer> longVersionedValue = new VersionedValue<>(REGISTER); - longVersionedValue.set(0, TEST_VALUE); + longVersionedValue.complete(0, TEST_VALUE); - REGISTER.moveRevision.accept(0L); + REGISTER.moveRevision(0L).join(); CompletableFuture<Integer> fut = longVersionedValue.get(1); @@ -206,16 +219,18 @@ public class VersionedValueTest { int incrementCount = 10; for (int i = 0; i < incrementCount; i++) { - longVersionedValue.update(1, previous -> ++previous, ex -> null); + longVersionedValue.update(1, (previous, e) -> completedFuture(++previous)); assertFalse(fut.isDone()); } - REGISTER.moveRevision.accept(1L); + REGISTER.moveRevision(1L).join(); assertTrue(fut.isDone()); assertEquals(TEST_VALUE + incrementCount, fut.get()); + + assertThrows(AssertionError.class, () -> longVersionedValue.update(1L, (i, t) -> completedFuture(null))); } /** @@ -225,28 +240,152 @@ public class VersionedValueTest { */ @Test public void testUpdatePredefined() throws Exception { - VersionedValue<Integer> longVersionedValue = new VersionedValue<>((b, r) -> { - }, REGISTER); + VersionedValue<Integer> longVersionedValue = new VersionedValue<>(REGISTER); CompletableFuture<Integer> fut = longVersionedValue.get(0); assertFalse(fut.isDone()); - longVersionedValue.update(0, previous -> { + longVersionedValue.update(0, (previous, e) -> { assertNull(previous); - return TEST_VALUE; - }, ex -> null); + return completedFuture(TEST_VALUE); + }); assertFalse(fut.isDone()); - REGISTER.moveRevision.accept(0L); + REGISTER.moveRevision(0L).join(); assertTrue(fut.isDone()); assertEquals(TEST_VALUE, fut.get()); } + /** + * Test asynchronous update closure. + */ + @Test + public void testAsyncUpdate() { + VersionedValue<Integer> vv = new VersionedValue<>(REGISTER); + + CompletableFuture<Integer> fut = new CompletableFuture<>(); + + vv.update(0L, (v, e) -> fut); + + CompletableFuture<Integer> vvFut = vv.get(0L); + + CompletableFuture<?> revFut = REGISTER.moveRevision(0L); + + assertFalse(fut.isDone()); + assertFalse(vvFut.isDone()); + assertFalse(revFut.isDone()); + + fut.complete(1); + + revFut.join(); + + assertTrue(vvFut.isDone()); + } + + /** + * Test the case when exception happens in updater. + */ + @Test + public void testExceptionOnUpdate() { + VersionedValue<Integer> vv = new VersionedValue<>(REGISTER, () -> 0); + + final int count = 4; + final int successfulCompletionsCount = count / 2; + + AtomicInteger actualSuccessfulCompletionsCount = new AtomicInteger(); + + final String exceptionMsg = "test msg"; + + for (int i = 0; i < count; i++) { + vv.update(0L, (v, e) -> { + if (e != null) { + return failedFuture(e); + } + + if (v == successfulCompletionsCount) { + throw new IgniteInternalException(exceptionMsg); + } + + actualSuccessfulCompletionsCount.incrementAndGet(); + + return completedFuture(++v); + }); + } + + AtomicReference<Throwable> exceptionRef = new AtomicReference<>(); + + vv.whenComplete((t, v, e) -> exceptionRef.set(e)); + + vv.complete(0L); + + assertThrowsWithCause(() -> vv.get(0L).join(), IgniteInternalException.class); + + assertEquals(exceptionMsg, exceptionRef.get().getMessage()); + assertEquals(successfulCompletionsCount, actualSuccessfulCompletionsCount.get()); + } + + /** + * Test with multiple versioned values and asynchronous completion. + */ + @Test + public void testAsyncMultiVv() { + final String registryName = "Registry"; + final String assignmentName = "Assignment"; + final String tableName = "T1_"; + + VersionedValue<Map<UUID, String>> tablesVv = new VersionedValue<>(f -> {}, HashMap::new); + VersionedValue<Map<UUID, String>> schemasVv = new VersionedValue<>(REGISTER, HashMap::new); + VersionedValue<Map<UUID, String>> assignmentsVv = new VersionedValue<>(REGISTER, HashMap::new); + + schemasVv.whenComplete((token, value, ex) -> tablesVv.complete(token)); + + BiFunction<Long, UUID, CompletableFuture<String>> schemaRegistry = + (token, uuid) -> schemasVv.get(token).thenApply(schemas -> schemas.get(uuid)); + + // Adding table. + long token = 0L; + UUID tableId = UUID.randomUUID(); + + CompletableFuture<String> tableFut = schemaRegistry.apply(token, tableId) + .thenCombine(assignmentsVv.get(token), (registry, assignments) -> tableName + registry + assignments.get(tableId)); + + tablesVv.update(token, (old, e) -> tableFut.thenApply(table -> { + Map<UUID, String> val = new HashMap<>(old); + + val.put(tableId, table); + + return val; + })); + + CompletableFuture<String> userFut = tablesVv.get(token).thenApply(map -> map.get(tableId)); + + schemasVv.update(token, (old, e) -> { + old.put(tableId, registryName); + + return completedFuture(old); + }); + + assignmentsVv.update(token, (old, e) -> { + old.put(tableId, assignmentName); + + return completedFuture(old); + }); + + assertFalse(tableFut.isDone()); + assertFalse(userFut.isDone()); + + REGISTER.moveRevision(token).join(); + + tableFut.join(); + + assertEquals(tableName + registryName + assignmentName, userFut.join()); + } + /** * Checks a behavior when {@link VersionedValue} has not initialized yet, but someone already tries to get a value. * @@ -264,7 +403,7 @@ public class VersionedValueTest { assertNull(longVersionedValue.latest()); - longVersionedValue.set(2, TEST_VALUE); + longVersionedValue.complete(2, TEST_VALUE); assertTrue(fut1.isDone()); assertTrue(fut2.isDone()); @@ -299,12 +438,10 @@ public class VersionedValueTest { public void checkDefaultValue(VersionedValue<Integer> vv, Integer expectedDefault) { assertEquals(expectedDefault, vv.latest()); - vv.update(0, a -> { + vv.update(0, (a, e) -> { assertEquals(expectedDefault, vv.latest()); - return a == null ? null : a + 1; - }, e -> { - throw new IgniteInternalException(e); + return completedFuture(a == null ? null : a + 1); } ); @@ -314,9 +451,9 @@ public class VersionedValueTest { assertFalse(f.isDone()); - vv.update(0, a -> a == null ? null : a + 1, e -> null); + vv.update(0, (a, e) -> completedFuture(a == null ? null : a + 1)); - REGISTER.moveRevision.accept(0L); + REGISTER.moveRevision(0L).join(); assertTrue(f.isDone()); @@ -324,17 +461,126 @@ public class VersionedValueTest { } /** - * Test revision register. + * Test {@link VersionedValue#whenComplete(IgniteTriConsumer)}. */ - private static class TestRevisionRegister implements Consumer<Consumer<Long>> { + @Test + public void testWhenComplete() { + VersionedValue<Integer> vv = new VersionedValue<>(REGISTER); + + AtomicInteger a = new AtomicInteger(); + AtomicInteger cntr = new AtomicInteger(-1); + + IgniteTriConsumer<Long, Integer, Throwable> listener = (t, v, e) -> { + if (e == null) { + a.set(v); + } else { + a.set(-1); + } + + cntr.incrementAndGet(); + }; + + vv.whenComplete(listener); + + // Test complete. + long token = 0; + + final long finalToken0 = token; + + vv.complete(token, TEST_VALUE); + + assertThrows(AssertionError.class, () -> vv.complete(finalToken0, 0)); + assertThrows(AssertionError.class, () -> vv.completeExceptionally(finalToken0, new Exception())); + + assertEquals(TEST_VALUE, a.get()); + assertEquals(token, cntr.get()); + + REGISTER.moveRevision(token).join(); + + // Test update. + token = 1; + + vv.update(token, (v, e) -> completedFuture(++v)); + + assertEquals(TEST_VALUE, a.get()); + + REGISTER.moveRevision(token).join(); + assertEquals(TEST_VALUE + 1, a.get()); + assertEquals(token, cntr.get()); + + // Test move revision. + token = 2; + + REGISTER.moveRevision(token).join(); + + assertEquals(TEST_VALUE + 1, a.get()); + assertEquals(token, cntr.get()); + + // Test complete exceptionally. + token = 3; + + final long finalToken3 = token; + + vv.completeExceptionally(token, new Exception()); + + assertThrows(AssertionError.class, () -> vv.complete(finalToken3, 0)); + assertThrows(AssertionError.class, () -> vv.completeExceptionally(finalToken3, new Exception())); + + assertEquals(-1, a.get()); + assertEquals(token, cntr.get()); + + REGISTER.moveRevision(token).join(); + + assertEquals(token, cntr.get()); + + // Test remove listener. + token = 4; + + vv.removeWhenComplete(listener); + + a.set(0); + + vv.complete(token, TEST_VALUE); + + assertEquals(0, a.get()); + assertEquals(token - 1, cntr.get()); + + REGISTER.moveRevision(token).join(); + } + + /** + * Test revision register. + */ + private static class TestRevisionRegister implements Consumer<Function<Long, CompletableFuture<?>>> { /** Revision consumer. */ - Consumer<Long> moveRevision; + List<Function<Long, CompletableFuture<?>>> moveRevisionList = new ArrayList<>(); /** {@inheritDoc} */ @Override - public void accept(Consumer<Long> consumer) { - moveRevision = consumer; + public void accept(Function<Long, CompletableFuture<?>> function) { + moveRevisionList.add(function); + } + + /** + * Clear list. + */ + public void clear() { + moveRevisionList.clear(); + } + + /** + * Move revision. + * + * @param revision Revision. + * @return Future for all listeners. + */ + public CompletableFuture<?> moveRevision(long revision) { + List<CompletableFuture<?>> futures = new ArrayList<>(); + + moveRevisionList.forEach(m -> futures.add(m.apply(revision))); + + return CompletableFuture.allOf(futures.toArray(new CompletableFuture[] {})); } } } diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java index db7c9c88f..49ec18799 100644 --- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java +++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java @@ -42,6 +42,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; +import java.util.function.Function; import java.util.function.IntFunction; import java.util.stream.IntStream; import org.apache.ignite.Ignite; @@ -246,13 +247,8 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest { modules.distributed().polymorphicSchemaExtensions() ); - Consumer<Consumer<Long>> registry = (c) -> { - clusterCfgMgr.configurationRegistry().listenUpdateStorageRevision(newStorageRevision -> { - c.accept(newStorageRevision); - - return CompletableFuture.completedFuture(null); - }); - }; + Consumer<Function<Long, CompletableFuture<?>>> registry = (c) -> clusterCfgMgr.configurationRegistry() + .listenUpdateStorageRevision(newStorageRevision -> c.apply(newStorageRevision)); DataStorageModules dataStorageModules = new DataStorageModules(ServiceLoader.load(DataStorageModule.class)); @@ -337,7 +333,7 @@ public class ItIgniteNodeRestartTest extends IgniteAbstractTest { // Deploy all registered watches because all components are ready and have registered their listeners. metaStorageMgr.deployWatches(); - configurationCatchUpFuture.join(); + assertThat(configurationCatchUpFuture, willCompleteSuccessfully()); log.info("Completed recovery on partially started node, last revision applied: " + lastRevision.get() + ", acceptableDifference: " + IgniteSystemProperties.getInteger(CONFIGURATION_CATCH_UP_DIFFERENCE_PROPERTY, 100) diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java index 6c08c585d..47259d32e 100644 --- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java +++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java @@ -28,6 +28,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.function.Consumer; +import java.util.function.Function; import org.apache.ignite.Ignite; import org.apache.ignite.IgnitionManager; import org.apache.ignite.client.handler.ClientHandlerModule; @@ -271,13 +272,8 @@ public class IgniteImpl implements Ignite { clusterSvc ); - Consumer<Consumer<Long>> registry = (c) -> { - clusterCfgMgr.configurationRegistry().listenUpdateStorageRevision(newStorageRevision -> { - c.accept(newStorageRevision); - - return CompletableFuture.completedFuture(null); - }); - }; + Consumer<Function<Long, CompletableFuture<?>>> registry = + c -> clusterCfgMgr.configurationRegistry().listenUpdateStorageRevision(c::apply); DataStorageModules dataStorageModules = new DataStorageModules( ServiceLoader.load(DataStorageModule.class, serviceProviderClassLoader) diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java index 66f3e8f80..f9f6dc3a8 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java @@ -26,6 +26,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; +import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -77,7 +78,7 @@ public class SqlQueryProcessor implements QueryProcessor { private final TableManager tableManager; - private final Consumer<Consumer<Long>> registry; + private final Consumer<Function<Long, CompletableFuture<?>>> registry; private final DataStorageManager dataStorageManager; @@ -101,7 +102,7 @@ public class SqlQueryProcessor implements QueryProcessor { /** Constructor. */ public SqlQueryProcessor( - Consumer<Consumer<Long>> registry, + Consumer<Function<Long, CompletableFuture<?>>> registry, ClusterService clusterSrvc, TableManager tableManager, DataStorageManager dataStorageManager, diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java index 5e3b94aef..ffbce8f98 100644 --- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java +++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/SqlSchemaManagerImpl.java @@ -17,14 +17,19 @@ package org.apache.ignite.internal.sql.engine.schema; +import static java.util.concurrent.CompletableFuture.completedFuture; +import static java.util.concurrent.CompletableFuture.failedFuture; + import java.util.Comparator; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArraySet; import java.util.function.Consumer; +import java.util.function.Function; import java.util.stream.Collectors; import org.apache.calcite.schema.SchemaPlus; import org.apache.calcite.tools.Frameworks; @@ -59,7 +64,7 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager { */ public SqlSchemaManagerImpl( TableManager tableManager, - Consumer<Consumer<Long>> registry + Consumer<Function<Long, CompletableFuture<?>>> registry ) { this.tableManager = tableManager; schemasVv = new VersionedValue<>(registry, HashMap::new); @@ -134,21 +139,22 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager { * @param causalityToken Causality token. */ public synchronized void onSchemaCreated(String schemaName, long causalityToken) { - Map<String, IgniteSchema> schemasMap = schemasVv.update( + CompletableFuture<Map<String, IgniteSchema>> schemasMapFut = schemasVv.update( causalityToken, - schemas -> { + (schemas, e) -> { + if (e != null) { + return failedFuture(e); + } + Map<String, IgniteSchema> res = new HashMap<>(schemas); res.putIfAbsent(schemaName, new IgniteSchema(schemaName)); - return res; - }, - e -> { - throw new IgniteInternalException(e); + return completedFuture(res); } ); - rebuild(causalityToken, schemasMap); + rebuild(causalityToken, schemasMapFut); } /** @@ -158,21 +164,22 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager { * @param causalityToken Causality token. */ public synchronized void onSchemaDropped(String schemaName, long causalityToken) { - Map<String, IgniteSchema> schemasMap = schemasVv.update( + CompletableFuture<Map<String, IgniteSchema>> schemasMapFut = schemasVv.update( causalityToken, - schemas -> { + (schemas, e) -> { + if (e != null) { + return failedFuture(e); + } + Map<String, IgniteSchema> res = new HashMap<>(schemas); res.remove(schemaName); - return res; - }, - e -> { - throw new IgniteInternalException(e); + return completedFuture(res); } ); - rebuild(causalityToken, schemasMap); + rebuild(causalityToken, schemasMapFut); } /** @@ -184,9 +191,13 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager { TableImpl table, long causalityToken ) { - Map<String, IgniteSchema> schemasMap = schemasVv.update( + CompletableFuture<Map<String, IgniteSchema>> schemasMapFut = schemasVv.update( causalityToken, - schemas -> { + (schemas, e) -> { + if (e != null) { + return failedFuture(e); + } + Map<String, IgniteSchema> res = new HashMap<>(schemas); IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new); @@ -195,28 +206,26 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager { schema.addTable(removeSchema(schemaName, table.name()), igniteTable); - tablesVv.update( + return tablesVv + .update( causalityToken, - tables -> { + (tables, ex) -> { + if (ex != null) { + return failedFuture(ex); + } + Map<UUID, IgniteTable> resTbls = new HashMap<>(tables); resTbls.put(igniteTable.id(), igniteTable); - return resTbls; - }, - e -> { - throw new IgniteInternalException(e); + return completedFuture(resTbls); } - ); - - return res; - }, - e -> { - throw new IgniteInternalException(e); + ) + .thenCompose(tables -> completedFuture(res)); } ); - rebuild(causalityToken, schemasMap); + rebuild(causalityToken, schemasMapFut); } /** @@ -240,8 +249,12 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager { String tableName, long causalityToken ) { - Map<String, IgniteSchema> schemasMap = schemasVv.update(causalityToken, - schemas -> { + CompletableFuture<Map<String, IgniteSchema>> schemasMapFut = schemasVv.update(causalityToken, + (schemas, e) -> { + if (e != null) { + return failedFuture(e); + } + Map<String, IgniteSchema> res = new HashMap<>(schemas); IgniteSchema schema = res.computeIfAbsent(schemaName, IgniteSchema::new); @@ -253,39 +266,43 @@ public class SqlSchemaManagerImpl implements SqlSchemaManager { if (table != null) { schema.removeTable(calciteTableName); - tablesVv.update(causalityToken, - tables -> { + return tablesVv + .update(causalityToken, + (tables, ex) -> { + if (ex != null) { + return failedFuture(ex); + } + Map<UUID, IgniteTable> resTbls = new HashMap<>(tables); resTbls.remove(table.id()); - return resTbls; - }, - e -> { - throw new IgniteInternalException(e); + return completedFuture(resTbls); } - ); + ) + .thenCompose(tables -> completedFuture(res)); } - return res; - }, - e -> { - throw new IgniteInternalException(e); + return completedFuture(res); } ); - rebuild(causalityToken, schemasMap); + rebuild(causalityToken, schemasMapFut); } - private void rebuild(long causalityToken, Map<String, IgniteSchema> schemas) { + private void rebuild(long causalityToken, CompletableFuture<Map<String, IgniteSchema>> schemasFut) { SchemaPlus newCalciteSchema = Frameworks.createRootSchema(false); newCalciteSchema.add("PUBLIC", new IgniteSchema("PUBLIC")); - schemas.forEach(newCalciteSchema::add); + schemasFut.join().forEach(newCalciteSchema::add); - calciteSchemaVv.update(causalityToken, s -> newCalciteSchema, e -> { - throw new IgniteInternalException(e); + calciteSchemaVv.update(causalityToken, (s, e) -> { + if (e != null) { + return failedFuture(e); + } + + return completedFuture(newCalciteSchema); }); listeners.forEach(SchemaUpdateListener::onSchemaUpdated); diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java index 79e23c0ea..99a5f60e4 100644 --- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java +++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java @@ -17,6 +17,7 @@ package org.apache.ignite.internal.sql.engine; +import static java.util.concurrent.CompletableFuture.allOf; import static org.apache.ignite.internal.schema.registry.SchemaRegistryImpl.INITIAL_SCHEMA_VERSION; import static org.apache.ignite.internal.testframework.IgniteTestUtils.await; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -37,9 +38,11 @@ import java.util.Arrays; import java.util.Collections; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.Flow; import java.util.function.Consumer; +import java.util.function.Function; import org.apache.ignite.configuration.ConfigurationValue; import org.apache.ignite.configuration.schemas.store.UnknownDataStorageConfigurationSchema; import org.apache.ignite.configuration.schemas.table.HashIndexConfigurationSchema; @@ -205,7 +208,7 @@ public class StopCalciteModuleTest { qryProc.start(); - testRevisionRegister.moveRevision.accept(0L); + testRevisionRegister.moveRevision.apply(0L).join(); var cursors = qryProc.queryAsync( "PUBLIC", @@ -259,18 +262,22 @@ public class StopCalciteModuleTest { /** * Test revision register. */ - private static class TestRevisionRegister implements Consumer<Consumer<Long>> { - + private static class TestRevisionRegister implements Consumer<Function<Long, CompletableFuture<?>>> { /** Revision consumer. */ - Consumer<Long> moveRevision; + Function<Long, CompletableFuture<?>> moveRevision; /** {@inheritDoc} */ @Override - public void accept(Consumer<Long> consumer) { + public void accept(Function<Long, CompletableFuture<?>> function) { if (moveRevision == null) { - moveRevision = consumer; + moveRevision = function; } else { - moveRevision = moveRevision.andThen(consumer); + Function<Long, CompletableFuture<?>> old = moveRevision; + + moveRevision = rev -> allOf( + old.apply(rev), + function.apply(rev) + ); } } } diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java index 542f2ba77..51249d8bb 100644 --- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java +++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java @@ -42,6 +42,7 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import java.util.function.Function; import org.apache.ignite.configuration.schemas.store.UnknownDataStorageConfigurationSchema; import org.apache.ignite.configuration.schemas.table.HashIndexConfigurationSchema; import org.apache.ignite.configuration.schemas.table.PartialIndexConfigurationSchema; @@ -138,7 +139,7 @@ public class MockedStructuresTest extends IgniteAbstractTest { private ConfigurationStorageRevisionListenerHolder fieldRevisionListenerHolder; /** Revision updater. */ - private Consumer<Consumer<Long>> revisionUpdater; + private Consumer<Function<Long, CompletableFuture<?>>> revisionUpdater; /** Tables configuration. */ @InjectConfiguration( @@ -202,15 +203,13 @@ public class MockedStructuresTest extends IgniteAbstractTest { /** Inner initialisation. */ @BeforeEach void before() throws Exception { - revisionUpdater = (Consumer<Long> consumer) -> { - consumer.accept(0L); + revisionUpdater = (Function<Long, CompletableFuture<?>> function) -> { + function.apply(0L).join(); fieldRevisionListenerHolder.listenUpdateStorageRevision(newStorageRevision -> { log.info("Notify about revision: {}", newStorageRevision); - consumer.accept(newStorageRevision); - - return CompletableFuture.completedFuture(null); + return function.apply(newStorageRevision); }); }; @@ -628,6 +627,14 @@ public class MockedStructuresTest extends IgniteAbstractTest { return completedFuture(raftGrpSrvcMock); }); + when(rm.updateRaftGroup(any(), any(), any(), any())).thenAnswer(mock -> { + RaftGroupService raftGrpSrvcMock = mock(RaftGroupService.class); + + when(raftGrpSrvcMock.leader()).thenReturn(new Peer(new NetworkAddress("localhost", 47500))); + + return completedFuture(raftGrpSrvcMock); + }); + when(ts.getByAddress(any(NetworkAddress.class))).thenReturn(new ClusterNode( UUID.randomUUID().toString(), "node0", diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java index ff085e11d..79dd5df1a 100644 --- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java +++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/schema/SqlSchemaManagerTest.java @@ -17,6 +17,7 @@ package org.apache.ignite.internal.sql.engine.exec.schema; +import static java.util.concurrent.CompletableFuture.allOf; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsString; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -30,8 +31,10 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.when; import java.util.UUID; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; +import java.util.function.Function; import org.apache.calcite.schema.Table; import org.apache.ignite.internal.schema.Column; import org.apache.ignite.internal.schema.NativeTypes; @@ -227,18 +230,17 @@ public class SqlSchemaManagerTest { /** * Test revision register. */ - private static class TestRevisionRegister implements Consumer<Consumer<Long>> { + private static class TestRevisionRegister implements Consumer<Function<Long, CompletableFuture<?>>> { AtomicLong token = new AtomicLong(-1); - /** Revision consumer. */ - private Consumer<Long> moveRevision; + private Function<Long, CompletableFuture<?>> moveRevision; /** * Moves forward token. */ void moveForward() { - moveRevision.accept(token.incrementAndGet()); + moveRevision.apply(token.incrementAndGet()).join(); } /** @@ -252,11 +254,16 @@ public class SqlSchemaManagerTest { /** {@inheritDoc} */ @Override - public void accept(Consumer<Long> consumer) { + public void accept(Function<Long, CompletableFuture<?>> function) { if (moveRevision == null) { - moveRevision = consumer; + moveRevision = function; } else { - moveRevision = moveRevision.andThen(consumer); + Function<Long, CompletableFuture<?>> old = moveRevision; + + moveRevision = rev -> allOf( + old.apply(rev), + function.apply(rev) + ); } } } diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java index 19209ea0d..32770d261 100644 --- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java +++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java @@ -18,6 +18,8 @@ package org.apache.ignite.internal.table.distributed; import static java.util.Collections.unmodifiableMap; +import static java.util.concurrent.CompletableFuture.completedFuture; +import static java.util.concurrent.CompletableFuture.failedFuture; import static org.apache.ignite.internal.configuration.util.ConfigurationUtil.getByInternalId; import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap; @@ -82,7 +84,6 @@ import org.apache.ignite.internal.util.ByteUtils; import org.apache.ignite.internal.util.IgniteObjectName; import org.apache.ignite.internal.util.IgniteSpinBusyLock; import org.apache.ignite.lang.IgniteException; -import org.apache.ignite.lang.IgniteInternalException; import org.apache.ignite.lang.IgniteLogger; import org.apache.ignite.lang.IgniteStringFormatter; import org.apache.ignite.lang.IgniteSystemProperties; @@ -163,7 +164,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp * @param dataStorageMgr Data storage manager. */ public TableManager( - Consumer<Consumer<Long>> registry, + Consumer<Function<Long, CompletableFuture<?>>> registry, TablesConfiguration tablesCfg, Loza raftMgr, BaselineManager baselineMgr, @@ -242,7 +243,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp new NodeStoppingException() ); - return CompletableFuture.failedFuture(new NodeStoppingException()); + return failedFuture(new NodeStoppingException()); } try { @@ -276,7 +277,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp new NodeStoppingException() ); - return CompletableFuture.failedFuture(new NodeStoppingException()); + return failedFuture(new NodeStoppingException()); } try { @@ -301,7 +302,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp */ private CompletableFuture<?> onUpdateAssignments(ConfigurationNotificationEvent<byte[]> assignmentsCtx) { if (!busyLock.enterBusy()) { - return CompletableFuture.failedFuture(new NodeStoppingException()); + return failedFuture(new NodeStoppingException()); } try { @@ -339,7 +340,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp ); } - return CompletableFuture.failedFuture(new NodeStoppingException()); + return failedFuture(new NodeStoppingException()); } try { @@ -365,7 +366,11 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp SchemaDescriptor schemaDescriptor = SchemaSerializerImpl.INSTANCE.deserialize((schemasCtx.newValue().schema())); - tablesByIdVv.update(causalityToken, tablesById -> { + tablesByIdVv.update(causalityToken, (tablesById, e) -> { + if (e != null) { + return failedFuture(e); + } + TableImpl table = tablesById.get(tblId); ((SchemaRegistryImpl) table.schemaView()).onSchemaRegistered(schemaDescriptor); @@ -374,10 +379,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp fireEvent(TableEvent.ALTER, new TableEventParameters(causalityToken, table), null); } - return tablesById; - }, th -> { - throw new IgniteInternalException(IgniteStringFormatter.format("Cannot create a schema for table" - + " [tableId={}, schemaVer={}]", tblId, schemaDescriptor.version()), th); + return completedFuture(tablesById); }); } @@ -422,7 +424,11 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp toAdd.removeAll(oldPartitionAssignment); // Create new raft nodes according to new assignments. - tablesByIdVv.update(causalityToken, tablesById -> { + tablesByIdVv.update(causalityToken, (tablesById, e) -> { + if (e != null) { + return failedFuture(e); + } + InternalTable internalTable = tablesById.get(tblId).internalTable(); try { @@ -441,14 +447,11 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp return null; }); - } catch (NodeStoppingException e) { - throw new AssertionError("Loza was stopped before Table manager", e); + } catch (NodeStoppingException ex) { + throw new AssertionError("Loza was stopped before Table manager", ex); } - return tablesById; - }, th -> { - throw new IgniteInternalException(IgniteStringFormatter.format("Cannot start RAFT group for table" - + " [tableId={}, part={}]", tblId, partId), th); + return completedFuture(tablesById); }); } @@ -522,24 +525,28 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp var table = new TableImpl(internalTable, schemaRegistry); - tablesVv.update(causalityToken, previous -> { + tablesVv.update(causalityToken, (previous, e) -> { + if (e != null) { + return failedFuture(e); + } + var val = new HashMap<>(previous); val.put(name, table); - return val; - }, th -> { - throw new IgniteInternalException(IgniteStringFormatter.format("Cannot create a table [name={}, id={}]", name, tblId), th); + return completedFuture(val); }); - tablesByIdVv.update(causalityToken, previous -> { + tablesByIdVv.update(causalityToken, (previous, e) -> { + if (e != null) { + return failedFuture(e); + } + var val = new HashMap<>(previous); val.put(tblId, table); - return val; - }, th -> { - throw new IgniteInternalException(IgniteStringFormatter.format("Cannot create a table [name={}, id={}]", name, tblId), th); + return completedFuture(val); }); CompletableFuture.allOf(tablesByIdVv.get(causalityToken), tablesVv.get(causalityToken)).thenRun(() -> { @@ -647,30 +654,32 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp raftMgr.stopRaftGroup(raftGroupName(tblId, p)); } - tablesVv.update(causalityToken, previousVal -> { + tablesVv.update(causalityToken, (previousVal, e) -> { + if (e != null) { + return failedFuture(e); + } + var map = new HashMap<>(previousVal); map.remove(name); - return map; - }, th -> { - throw new IgniteInternalException(IgniteStringFormatter.format("Cannot drop a table [name={}, id={}]", name, tblId), - th); + return completedFuture(map); }); AtomicReference<TableImpl> tableHolder = new AtomicReference<>(); - tablesByIdVv.update(causalityToken, previousVal -> { + tablesByIdVv.update(causalityToken, (previousVal, e) -> { + if (e != null) { + return failedFuture(e); + } + var map = new HashMap<>(previousVal); TableImpl table = map.remove(tblId); tableHolder.set(table); - return map; - }, th -> { - throw new IgniteInternalException(IgniteStringFormatter.format("Cannot drop a table [name={}, id={}]", name, tblId), - th); + return completedFuture(map); }); TableImpl table = tableHolder.get(); diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/TableManagerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/TableManagerTest.java index 0c842aca5..a1b6b3597 100644 --- a/modules/table/src/test/java/org/apache/ignite/internal/table/TableManagerTest.java +++ b/modules/table/src/test/java/org/apache/ignite/internal/table/TableManagerTest.java @@ -45,6 +45,7 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Phaser; import java.util.function.Consumer; +import java.util.function.Function; import org.apache.ignite.configuration.NamedListView; import org.apache.ignite.configuration.schemas.store.UnknownDataStorageConfigurationSchema; import org.apache.ignite.configuration.schemas.table.HashIndexConfigurationSchema; @@ -156,7 +157,7 @@ public class TableManagerTest extends IgniteAbstractTest { private ConfigurationStorageRevisionListenerHolder fieldRevisionListenerHolder; /** Revision updater. */ - private Consumer<Consumer<Long>> revisionUpdater; + private Consumer<Function<Long, CompletableFuture<?>>> revisionUpdater; /** Tables configuration. */ @InjectConfiguration( @@ -192,15 +193,13 @@ public class TableManagerTest extends IgniteAbstractTest { /** Before all test scenarios. */ @BeforeEach void before() { - revisionUpdater = (Consumer<Long> consumer) -> { - consumer.accept(0L); + revisionUpdater = (Function<Long, CompletableFuture<?>> function) -> { + function.apply(0L).join(); fieldRevisionListenerHolder.listenUpdateStorageRevision(newStorageRevision -> { log.info("Notify about revision: {}", newStorageRevision); - consumer.accept(newStorageRevision); - - return CompletableFuture.completedFuture(null); + return function.apply(newStorageRevision); }); };