diff --git a/modules/commons/src/main/java/org/apache/ignite/internal/util/CommonUtils.java b/modules/commons/src/main/java/org/apache/ignite/internal/util/CommonUtils.java index cb2c770007fa3..7d6f1940e9e88 100644 --- a/modules/commons/src/main/java/org/apache/ignite/internal/util/CommonUtils.java +++ b/modules/commons/src/main/java/org/apache/ignite/internal/util/CommonUtils.java @@ -68,6 +68,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteClientDisconnectedException; import org.apache.ignite.IgniteCommonsSystemProperties; @@ -231,6 +232,12 @@ public abstract class CommonUtils { /** Default client connector port range. */ public static final int DFLT_PORT_RANGE = 100; + /** + * Default timeout in milliseconds to wait for a sub-routines, runnables to finish on {@link IgniteKernal#stop(boolean)} + * or other stop or cancelation requests. + */ + public static int DFLT_WAIT_TO_STOP_TIMOEUT = 60_000; + /** * Short date format pattern for log messages in "quiet" mode. * Only time is included since we don't expect "quiet" mode to be used @@ -2468,43 +2475,82 @@ public static void cancel(Iterable ws) { } /** - * Joins runnable. + * Joins grid worker with a timeout. Logs exceptable failures. * - * @param w Worker to join. - * @param log The logger to possible exception. - * @return {@code true} if worker has not been interrupted, {@code false} if it was interrupted. + * @param worker Worker to join. + * @param timeout Timeout to join in milliseconds. If negative, ignored. + * @param log The logger for exceptable failures. + * @return {@code True} if {@code worker} has been successfully stopped, wasn't interrupted or timeouted. {@code False} + * if {@code worker} was interrupted or timeouted. */ - public static boolean join(@Nullable GridWorker w, @Nullable IgniteLogger log) { - if (w != null) - try { - w.join(); - } - catch (InterruptedException ignore) { - warn(log, "Got interrupted while waiting for completion of runnable: " + w); + public static boolean join(@Nullable GridWorker worker, long timeout, IgniteLogger log) { + assert log != null; - Thread.currentThread().interrupt(); + if (worker == null) + return true; - return false; - } + try { + worker.join(timeout); + } + catch (InterruptedException ignore) { + warn(log, "Got interrupted while waiting for completion of grid worker: " + worker); + + Thread.currentThread().interrupt(); + + return false; + } + catch (TimeoutException te) { + warn(log, "The timeout expired while waiting for completion of grid worker: " + worker); + + return false; + } return true; } /** - * Joins given collection of runnables. + * Joins grid worker with unlimited waiting. Logs exceptable failures. + * + * @param worker Worker to join. + * @param log The logger for exceptable failures. + * @return {@code True} if {@code worker} has been successfully stopped, wasn't interrupted or timeouted. {@code False} + * if {@code worker} was interrupted or timeouted. + */ + public static boolean join(@Nullable GridWorker worker, IgniteLogger log) { + return join(worker, -1L, log); + } + + /** + * Joins given collection of runnables with a timoeut. Logs exceptable failures. * * @param ws Collection of workers to join. + * @param timeout Timeout to join in milliseconds. If negative, ignored. * @param log The logger to possible exceptions. * @return {@code true} if none of the worker have been interrupted, * {@code false} if at least one was interrupted. */ - public static boolean join(Iterable ws, IgniteLogger log) { + public static boolean join(@Nullable Iterable ws, long timeout, IgniteLogger log) { + if (ws == null) + return true; + boolean retval = true; - if (ws != null) - for (GridWorker w : ws) - if (!join(w, log)) + if (timeout < 0) { + for (GridWorker w : ws) { + if (!join(w, -1L, log)) retval = false; + } + } + else { + long timeThresholdNs = System.nanoTime() + millisToNanos(timeout); + + for (GridWorker w : ws) { + long timeout0 = Math.max(0L, nanosToMillis(timeThresholdNs - System.nanoTime())); + + if (!join(w, timeout0, log)) + retval = false; + } + } return retval; } diff --git a/modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorker.java b/modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorker.java index 414b8dac3c704..b5e745da4f1d2 100644 --- a/modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorker.java +++ b/modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorker.java @@ -19,6 +19,7 @@ import java.util.concurrent.Executor; import java.util.concurrent.Future; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLongFieldUpdater; import org.apache.ignite.IgniteInterruptedException; @@ -227,20 +228,43 @@ public void cancel() { } /** - * Joins this runnable. + * Joins this runnable with a timeout. * - * @throws InterruptedException Thrown in case of interruption. + * @param timeout Operation timeout in milliseconds. If negative, ignored. + * @throws InterruptedException In case of the interruption. + * @throws TimeoutException In case of expired {@code timeout} to join. */ - public void join() throws InterruptedException { - if (log.isDebugEnabled()) - log.debug("Joining grid runnable: " + this); + public void join(long timeout) throws InterruptedException, TimeoutException { + long timeThresholdNs = timeout < 0L ? -1L : System.nanoTime() + CommonUtils.millisToNanos(timeout); + + if (log.isDebugEnabled()) { + if (timeout >= 0L) + log.debug("Joining grid runnable: " + this + " with the Wtimeout: " + timeout + "ms."); + else + log.debug("Joining grid runnable: " + this + '.'); + } if ((runner == null && isCancelled.get()) || finished) return; - synchronized (mux) { - while (!finished) - mux.wait(); + while (!finished) { + if (timeout >= 0L) { + long leftMs = CommonUtils.nanosToMillis(timeThresholdNs - System.nanoTime()); + + if (leftMs < 1L) + throw new TimeoutException("The timeout has expired while waiting to join."); + + synchronized (mux) { + if (!finished) + mux.wait(leftMs); + } + } + else { + synchronized (mux) { + if (!finished) + mux.wait(); + } + } } } diff --git a/modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorkerPool.java b/modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorkerPool.java index 51b6a97bd3265..6cd88de2add59 100644 --- a/modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorkerPool.java +++ b/modules/commons/src/main/java/org/apache/ignite/internal/util/worker/GridWorkerPool.java @@ -93,20 +93,36 @@ public void execute(final GridWorker w) throws IgniteCheckedException { } /** - * Waits for all workers to finish. + * Waits for all workers to finish with a timeout. Logs failures if {@link #log} is not {@code null}. * - * @param cancel Flag to indicate whether workers should be cancelled - * before waiting for them to finish. + * @param cancel Flag to indicate whether workers should be cancelled before waiting for them to finish. + * @param timeout Timeout to join in milliseconds. If negative, ignored. */ - public void join(boolean cancel) { + public void join(boolean cancel, long timeout) { + long timeThresholdNs = timeout < 0L ? -1L : System.nanoTime() + CommonUtils.millisToNanos(timeout); + + if (cancel) { + for (GridWorker worker : workers) { + try { + if (cancel) + CommonUtils.cancel(worker); + } + catch (Throwable e) { + if (log != null) + log.warning("Failed to cancel grid worker [" + worker.name() + ']', e); + } + } + } + for (GridWorker worker : workers) { try { - if (cancel) - CommonUtils.cancel(worker); - - CommonUtils.join(worker, log); + CommonUtils.join( + worker, + timeout < 0L ? -1L : CommonUtils.nanosToMillis(timeThresholdNs - System.nanoTime()), + log + ); } - catch (Exception e) { + catch (Throwable e) { if (log != null) log.warning("Failed to stop grid worker [" + worker.name() + ']', e); } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java index 009bbfa0dc208..2817381bcde9c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java @@ -96,6 +96,7 @@ import org.apache.ignite.internal.thread.context.OperationContext; import org.apache.ignite.internal.thread.context.Scope; import org.apache.ignite.internal.thread.context.function.OperationContextAwareWrapper; +import org.apache.ignite.internal.util.CommonUtils; import org.apache.ignite.internal.util.GridAtomicLong; import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashMap; import org.apache.ignite.internal.util.GridSpinBusyLock; @@ -1683,9 +1684,10 @@ private static String nodeDescription(ClusterNode node) { // Stop segment checker. if (segChecker != null) { - segChecker.cancel(); + if (cancel) + segChecker.cancel(); - U.join(segChecker, log); + U.join(segChecker, 0L, log); } if (!locJoin.isDone()) @@ -1700,12 +1702,11 @@ private static String nodeDescription(ClusterNode node) { getSpi().setListener(null); U.cancel(discoEvtHnd); - - U.join(discoEvtHnd, log); - U.cancel(discoMsgNotifier); - U.join(discoMsgNotifier, log); + U.join(discoMsgNotifier, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); + // Event processing might have crusial routines better to wait for. Or requiring to stop (process) manually. + U.join(discoEvtHnd, log); // Stop SPI itself. stopSpi(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java index 021fdd0408d48..16d1f779b9d7c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java @@ -5016,7 +5016,7 @@ else if (newSysTtl == CU.TTL_ZERO) { interceptorVal = cctx.config().getInterceptor().onBeforePut(interceptEntry, updated0); } catch (Throwable e) { - throw new IgniteCheckedException(e); + throw new IgniteCheckedException("Cache's interceptor failed in 'onBeforePut'.", e); } wasIntercepted = true; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java index 9fe3462af0546..23ce1f06a59f8 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java @@ -466,7 +466,10 @@ public GridCacheVersion mappedVersion(GridCacheVersion from) { public void onStop() { stopping = true; - cancelClientFutures(stopError()); + var stopErr = stopError(); + + cancelClientFutures(stopErr); + cancelFuturesWithException(stopErr, dataStreamerFuts); } /** {@inheritDoc} */ @@ -496,7 +499,7 @@ private void cancelFuturesWithException( try { ((GridFutureAdapter)fut).onDone(err); } - catch (Exception e) { + catch (Throwable e) { U.warn(log, "Failed to complete future on node stop (will ignore): " + fut, e); } } @@ -612,6 +615,8 @@ public GridFutureAdapter addDataStreamerFuture(AffinityTopologyVersion topVer) { assert add; + onFutureAdded(fut); + return fut; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java index 621cabfad03fa..b0dfaf5ad7785 100755 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java @@ -990,7 +990,7 @@ public T afterAsyncCompletion(IgniteOutClosure closure) { * Cancels async tasks. */ public void awaitAsyncTaskCompletion(boolean cancel) { - workerPool.join(cancel); + workerPool.join(cancel, -1L); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java index 0276827b6cc45..d1f3bb1e44e19 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobProcessor.java @@ -495,8 +495,8 @@ public GridJobProcessor(GridKernalContext ctx) { } } - U.join(activeJobs.values(), log); - U.join(cancelledJobs.values(), log); + U.join(activeJobs.values(), -1L, log); + U.join(cancelledJobs.values(), -1L, log); // Ignore topology changes. ctx.event().removeLocalEventListener(discoLsnr); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java index 808437eb1719a..9a71e6d25b56f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/odbc/ClientListenerProcessor.java @@ -49,6 +49,7 @@ import org.apache.ignite.internal.processors.odbc.odbc.OdbcConnectionContext; import org.apache.ignite.internal.systemview.ClientConnectionAttributeViewWalker; import org.apache.ignite.internal.systemview.ClientConnectionViewWalker; +import org.apache.ignite.internal.util.CommonUtils; import org.apache.ignite.internal.util.GridSpinBusyLock; import org.apache.ignite.internal.util.HostAndPortRange; import org.apache.ignite.internal.util.nio.GridNioAsyncNotifyFilter; @@ -529,7 +530,7 @@ else if (connCtx.managementClient()) { execSvc = null; - mgmtPool.join(cancel); + mgmtPool.join(cancel, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT); mgmtPool = null; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentManager.java index 690b4b02001f2..ce0b201c5e104 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentManager.java @@ -46,6 +46,7 @@ import org.apache.ignite.internal.thread.context.OperationContext; import org.apache.ignite.internal.thread.context.Scope; import org.apache.ignite.internal.thread.context.function.OperationContextAwareWrapper; +import org.apache.ignite.internal.util.CommonUtils; import org.apache.ignite.internal.util.GridSpinBusyLock; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.internal.util.worker.GridWorker; @@ -145,7 +146,7 @@ void stopProcessing(IgniteCheckedException stopErr) { U.cancel(depTaskHandler); - U.join(depTaskHandler, log); + U.join(depTaskHandler, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); depTaskHandler.clearQueue(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java index 4386f8313e3c3..0187a784a6dda 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskProcessor.java @@ -308,7 +308,7 @@ private IgniteClientDisconnectedCheckedException disconnectedError(@Nullable Ign } } - U.join(tasks.values(), log); + U.join(tasks.values(), -1L, log); } // Remove discovery and message listeners. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java index 5b87447a84813..46e91c1345421 100755 --- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java @@ -3151,60 +3151,75 @@ public static void interrupt(Iterable workers) { } /** - * Waits for completion of a given thread. If thread is {@code null} then - * this method returns immediately returning {@code true} + * Waits for completion of a given thread with a timeout. * * @param t Thread to join. + * @param timeout Join timeout. * @param log Logger for logging errors. - * @return {@code true} if thread has finished, {@code false} otherwise. + * @return {@code True} if thread has finished ot if {@code t} is {@code null}. {@code False} otherwise. */ - public static boolean join(@Nullable Thread t, @Nullable IgniteLogger log) { - return join(t, log, 0); + public static boolean join(@Nullable Thread t, long timeout, IgniteLogger log) { + assert log != null; + + if (t == null) + return true; + + try { + t.join(timeout); + + return !t.isAlive(); + } + catch (InterruptedException ignore) { + warn(log, "Got interrupted while waiting for completion of a thread: " + t); + + Thread.currentThread().interrupt(); + + return false; + } } /** - * Waits for completion of a given thread. If thread is {@code null} then - * this method returns immediately returning {@code true} + * Waits for completion of a given thread with unlimited waiting. * * @param t Thread to join. * @param log Logger for logging errors. - * @param timeout Join timeout. - * @return {@code true} if thread has finished, {@code false} otherwise. + * @return {@code True} if thread has finished ot if {@code t} is {@code null}. {@code False} otherwise. */ - public static boolean join(@Nullable Thread t, @Nullable IgniteLogger log, long timeout) { - if (t != null) { - try { - t.join(timeout); - - return !t.isAlive(); - } - catch (InterruptedException ignore) { - warn(log, "Got interrupted while waiting for completion of a thread: " + t); - - Thread.currentThread().interrupt(); - - return false; - } - } - - return true; + public static boolean join(@Nullable Thread t, IgniteLogger log) { + return join(t, 0L, log); } /** - * Waits for completion of a given threads. If thread is {@code null} then + * Waits for completion of a given threads with a timoeut. If thread is {@code null} then * this method returns immediately returning {@code true} * - * @param workers Thread to join. + * @param threads Threads to join. + * @param timeout Operation timeout in milliseconds. * @param log Logger for logging errors. * @return {@code true} if thread has finished, {@code false} otherwise. */ - public static boolean joinThreads(Iterable workers, @Nullable IgniteLogger log) { + public static boolean joinThreads(Iterable threads, long timeout, @Nullable IgniteLogger log) { + if (threads == null) + return true; + boolean retval = true; - if (workers != null) - for (Thread worker : workers) + if (timeout < 0L) { + for (Thread worker : threads) { if (!join(worker, log)) retval = false; + } + } + else { + long timeThresholsNs = System.nanoTime(); + + for (Thread worker : threads) { + long timeout0 = Math.max(0, nanosToMillis(timeThresholsNs - System.nanoTime())); + + if (!join(worker, timeout0, log)) + retval = false; + } + } return retval; } @@ -4969,20 +4984,25 @@ public static void awaitQuiet(CyclicBarrier barrier) { } /** - * Joins worker. + * Joins grid worker with unlimited waiting. * - * @param w Worker. - * @throws IgniteInterruptedCheckedException Wrapped {@link InterruptedException}. + * @param worker Grid worker. + * @throws IgniteInterruptedCheckedException in case of {@link InterruptedException} or {@link TimeoutException}. */ - public static void join(GridWorker w) throws IgniteInterruptedCheckedException { + public static void join(@Nullable GridWorker worker) throws IgniteInterruptedCheckedException { + if (worker == null) + return; + try { - if (w != null) - w.join(); + worker.join(-1L); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - throw new IgniteInterruptedCheckedException(e); + throw new IgniteInterruptedCheckedException("Interrupted while waiting for grid worker " + worker, e); + } + catch (TimeoutException te) { + assert false : "TimeoutException is not expected if not actual timeout is set."; } } @@ -7218,21 +7238,27 @@ public static void notifyListeners(T t, Collection> lsnrs, Ignit * @param cancel Wheter should cancel workers. * @param log Logger. */ - public static void awaitForWorkersStop( - Collection workers, - boolean cancel, - @Nullable IgniteLogger log - ) { + public static void awaitForWorkersStop(Collection workers, boolean cancel, @Nullable IgniteLogger log) { + if (cancel) { + for (GridWorker worker : workers) { + try { + if (cancel) + worker.cancel(); + } + catch (Throwable e) { + if (log != null) + log.warning("Failed to cancel grid worker, worker=[" + worker + "], error: " + e.getMessage()); + } + } + } + for (GridWorker worker : workers) { try { - if (cancel) - worker.cancel(); - - worker.join(); + worker.join(-1L); } - catch (Exception e) { + catch (Throwable e) { if (log != null) - log.warning("Failed to cancel grid runnable [" + worker.toString() + "]: " + e.getMessage()); + log.warning("Failed to wait for grid worker to finish, worker=[" + worker + "], error: " + e.getMessage()); } } } diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java index b73d65497a8d8..c99941bc1f557 100755 --- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java @@ -46,6 +46,7 @@ import org.apache.ignite.internal.processors.failure.FailureProcessor; import org.apache.ignite.internal.processors.metric.impl.MetricUtils; import org.apache.ignite.internal.processors.resource.GridResourceProcessor; +import org.apache.ignite.internal.util.CommonUtils; import org.apache.ignite.internal.util.future.GridFinishedFuture; import org.apache.ignite.internal.util.future.IgniteFutureImpl; import org.apache.ignite.internal.util.nio.GridCommunicationClient; @@ -842,7 +843,7 @@ public boolean spiContextInitialized() { if (conStateHnd != null) { conStateHnd.stop(); U.cancel(conStateHnd); - U.join(conStateHnd, log); + U.join(conStateHnd, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); } if (srvLsnr != null) diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java index 53b1dbfc93556..a8657c85b0947 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java @@ -66,6 +66,7 @@ import org.apache.ignite.internal.IgniteNodeAttributes; import org.apache.ignite.internal.managers.discovery.DiscoveryServerOnlyCustomMessage; import org.apache.ignite.internal.thread.context.Scope; +import org.apache.ignite.internal.util.CommonUtils; import org.apache.ignite.internal.util.future.GridFutureAdapter; import org.apache.ignite.internal.util.tostring.GridToStringExclude; import org.apache.ignite.internal.util.typedef.F; @@ -353,7 +354,7 @@ class ClientImpl extends TcpDiscoveryImpl { U.join(sockWriter, log); // SocketReader may loose interruption, this hack is made to overcome that case. - while (!U.join(sockReader, log, 200)) + while (!U.join(sockReader, 200, log)) U.interrupt(sockReader); executorSrvc.shutdownNow(); @@ -452,10 +453,10 @@ else if (state == DISCONNECTED) { U.interrupt(sockReader); if (msgWorker != null) - U.join(msgWorker.runner(), log); + U.join(msgWorker.runner(), CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); - U.join(sockWriter, log); - U.join(sockReader, log); + U.join(sockWriter, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); + U.join(sockReader, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); leaveLatch.countDown(); joinLatch.countDown(); diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java index c97661427c1fb..63db786f5cec1 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java @@ -91,6 +91,7 @@ import org.apache.ignite.internal.processors.security.SecurityContext; import org.apache.ignite.internal.thread.context.Scope; import org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor; +import org.apache.ignite.internal.util.CommonUtils; import org.apache.ignite.internal.util.GridBoundedLinkedHashSet; import org.apache.ignite.internal.util.GridConcurrentHashSet; import org.apache.ignite.internal.util.IgniteUtils; @@ -619,14 +620,13 @@ else if (log.isInfoEnabled()) { tmp = U.arrayList(readers); } - U.interrupt(tmp); - U.joinThreads(tmp, log); - + U.cancel(msgWorker); U.interrupt(ipFinderCleaner); - U.join(ipFinderCleaner, log); + U.interrupt(tmp); - U.cancel(msgWorker); - U.join(msgWorker, log); + U.joinThreads(tmp, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); + U.join(ipFinderCleaner, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); + U.join(msgWorker, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); for (ClientMessageWorker clientWorker : clientMsgWorkers.values()) { if (clientWorker != null) { @@ -1975,7 +1975,7 @@ private void clearNodeAddedMessage(TcpDiscoveryAbstractMessage msg) { } U.interrupt(tmp); - U.joinThreads(tmp, log); + U.joinThreads(tmp, 0L, log); U.cancel(msgWorker); U.join(msgWorker, log); @@ -6446,7 +6446,7 @@ public void stop() { U.close(srvrSock, log); - U.join(TcpServer.this, log); + U.join(this, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); } } diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java index a2c913160b21b..784fe2818daa7 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java @@ -35,6 +35,7 @@ import org.apache.ignite.IgniteLogger; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.util.CommonUtils; import org.apache.ignite.internal.util.tostring.GridToStringExclude; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.T2; @@ -681,7 +682,7 @@ private T2, Boolean> requestAddresses(InetAddress U.interrupt(addrSnd); for (AddressSender addrSnd : addrSnds) - U.join(addrSnd, log); + U.join(addrSnd, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/MutallyDependentCacheUpdateFailureAtNodesStoppageTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/MutallyDependentCacheUpdateFailureAtNodesStoppageTest.java new file mode 100644 index 0000000000000..4f10455f7a09a --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/MutallyDependentCacheUpdateFailureAtNodesStoppageTest.java @@ -0,0 +1,189 @@ +/* + * 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.internal.processors.cache.distributed; + +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.CountDownLatch; +import javax.cache.Cache; +import org.apache.ignite.IgniteException; +import org.apache.ignite.Ignition; +import org.apache.ignite.cache.CacheInterceptor; +import org.apache.ignite.client.IgniteClient; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.ClientConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.TestRecordingCommunicationSpi; +import org.apache.ignite.lang.IgniteBiTuple; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.jetbrains.annotations.Nullable; +import org.junit.Test; + +import static java.util.Collections.singletonMap; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC; +import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; +import static org.apache.ignite.internal.processors.cache.distributed.GridCacheModuloAffinityFunction.IDX_ATTR; + +/** */ +public class MutallyDependentCacheUpdateFailureAtNodesStoppageTest extends GridCommonAbstractTest { + /** */ + public static final int NODE_1_FIRST_KEY = 1; + + /** */ + public static final int NODE_1_SECOND_KEY = 4; + + /** */ + public static final int NODE_2_FIRST_KEY = 2; + + /** */ + public static final int NODE_2_SECOND_KEY = 5; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + return super.getConfiguration(igniteInstanceName) + .setCommunicationSpi(new TestRecordingCommunicationSpi()) + .setUserAttributes(singletonMap(IDX_ATTR, getTestIgniteInstanceIndex(igniteInstanceName))); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @Test + public void testCacheEntriesProcessingFailureCausedByNodeStop() throws Exception { + startGridsMultiThreaded(3); + + TestInterceptor.putStartedLatch = new CountDownLatch(2); + TestInterceptor.putUnblockedLatch = new CountDownLatch(1); + + grid(0).createCache(createTestCacheConfiguration()); + + try ( + IgniteClient cli1 = Ignition.startClient(new ClientConfiguration().setClusterDiscoveryEnabled(false) + .setAddresses("127.0.0.1:10801")); + IgniteClient cli2 = Ignition.startClient(new ClientConfiguration().setClusterDiscoveryEnabled(false) + .setAddresses("127.0.0.1:10802")) + ) { + IgniteInternalFuture putFut1 = GridTestUtils.runAsync(() -> cli1.cache(DEFAULT_CACHE_NAME) + .putAll(createKeysForNode(2))); + IgniteInternalFuture putFut2 = GridTestUtils.runAsync(() -> cli2.cache(DEFAULT_CACHE_NAME) + .putAll(createKeysForNode(1))); + + assertTrue(TestInterceptor.putStartedLatch.await(getTestTimeout(), MILLISECONDS)); + + IgniteInternalFuture stopFut1 = GridTestUtils.runAsync(() -> stopGrid(1)); + IgniteInternalFuture stopFut2 = GridTestUtils.runAsync(() -> stopGrid(2)); + + try { + TestInterceptor.putUnblockedLatch.countDown(); + + stopFut1.get(getTestTimeout()); + stopFut2.get(getTestTimeout()); + + putFut1.get(getTestTimeout()); + putFut2.get(getTestTimeout()); + } + catch (Throwable e) { + assertTrue(e.getMessage().contains("Connection refused")); + } + } + } + + /** */ + private CacheConfiguration createTestCacheConfiguration() { + return new CacheConfiguration() + .setName(DEFAULT_CACHE_NAME) + .setAtomicityMode(ATOMIC) + .setWriteSynchronizationMode(FULL_SYNC) + .setBackups(2) + .setAffinity(new GridCacheModuloAffinityFunction(3, 2)) + .setInterceptor(new TestInterceptor()); + } + + /** */ + private Map createKeysForNode(int nodeIdx) { + Map data = new TreeMap<>(); + + if (nodeIdx == 2) { + data.put(NODE_2_FIRST_KEY, 2); + data.put(NODE_2_SECOND_KEY, 5); + } + else { + data.put(NODE_1_FIRST_KEY, 1); + data.put(NODE_1_SECOND_KEY, 4); + } + + return data; + } + + /** */ + private static final class TestInterceptor implements CacheInterceptor { + /** */ + private static CountDownLatch putStartedLatch; + + /** */ + private static CountDownLatch putUnblockedLatch; + + /** {@inheritDoc} */ + @Override public @Nullable Integer onGet(Integer key, @Nullable Integer val) { + return val; + } + + /** {@inheritDoc} */ + @Override public @Nullable Integer onBeforePut(Cache.Entry entry, Integer newVal) { + if (entry.getKey() == NODE_1_FIRST_KEY || entry.getKey() == NODE_2_FIRST_KEY) { + putStartedLatch.countDown(); + + try { + assertTrue(putUnblockedLatch.await(10000, MILLISECONDS)); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + + throw new IgniteException(e); + } + } + else + throw new RuntimeException("Test failure in interceptor"); + + return newVal; + } + + /** {@inheritDoc} */ + @Override public void onAfterPut(Cache.Entry entry) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public @Nullable IgniteBiTuple onBeforeRemove(Cache.Entry entry) { + return null; + } + + /** {@inheritDoc} */ + @Override public void onAfterRemove(Cache.Entry entry) { + // No-op. + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestSafeThreadFactory.java b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestSafeThreadFactory.java index 8c47eff60eb0a..0bbd2ee2b79de 100644 --- a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestSafeThreadFactory.java +++ b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestSafeThreadFactory.java @@ -194,7 +194,7 @@ static void stopAllThreads(IgniteLogger log) { U.interrupt(all); - U.joinThreads(all, log); + U.joinThreads(all, 0L, log); Iterator it = all.iterator(); diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java index dd18bf7f1c4db..a45050771a450 100755 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite10.java @@ -93,6 +93,7 @@ import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheMessageWriteTimeoutTest; import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheSystemTransactionsSelfTest; import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheTxMessageRecoveryTest; +import org.apache.ignite.internal.processors.cache.distributed.MutallyDependentCacheUpdateFailureAtNodesStoppageTest; import org.apache.ignite.internal.processors.cache.distributed.dht.GridCacheAtomicNearCacheSelfTest; import org.apache.ignite.internal.processors.cache.distributed.dht.GridCacheColocatedTxExceptionSelfTest; import org.apache.ignite.internal.processors.cache.distributed.dht.GridCacheGlobalLoadTest; @@ -133,6 +134,7 @@ public static List> suite(Collection ignoredTests) { GridTestUtils.addTestIfNeeded(suite, IgniteCacheTransactionalStopBusySelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCacheAtomicNearCacheSelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, CacheAtomicNearUpdateTopologyChangeTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, MutallyDependentCacheUpdateFailureAtNodesStoppageTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, CacheTxNearUpdateTopologyChangeTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCachePartitionedStorePutSelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCacheOffHeapMultiThreadedUpdateSelfTest.class, ignoredTests); diff --git a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioAsyncNotifyFilter.java b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioAsyncNotifyFilter.java index 4a5889f17dfb3..c5bac30cbb6dc 100644 --- a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioAsyncNotifyFilter.java +++ b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioAsyncNotifyFilter.java @@ -17,9 +17,12 @@ package org.apache.ignite.internal.util.nio; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteInterruptedException; import org.apache.ignite.IgniteLogger; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.util.CommonUtils; @@ -32,13 +35,13 @@ */ public class GridNioAsyncNotifyFilter extends GridNioFilterAdapter { /** Logger. */ - private IgniteLogger log; + private final IgniteLogger log; /** Worker pool. */ - private GridWorkerPool workerPool; + private final GridWorkerPool workerPool; /** Ignite instance name. */ - private String igniteInstanceName; + private final String igniteInstanceName; /** * Assigns filter name to a filter. @@ -58,7 +61,29 @@ public GridNioAsyncNotifyFilter(String igniteInstanceName, Executor exec, Ignite /** {@inheritDoc} */ @Override public void stop() { - workerPool.join(false); + CountDownLatch launchedLatch = new CountDownLatch(1); + + var stoppingT = new Thread( + () -> { + launchedLatch.countDown(); + + workerPool.join(true, -1L); + }, + getClass().getSimpleName() + "-stoppingDaemon" + ); + + stoppingT.setDaemon(true); + + stoppingT.start(); + + try { + launchedLatch.await(CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + + throw new IgniteInterruptedException("Filed to wait for " + getClass().getName() + "'s stopping thread launch.", e); + } } /** {@inheritDoc} */ diff --git a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java index a067277648d35..5a3f4f676393f 100644 --- a/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java +++ b/modules/nio/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java @@ -512,10 +512,10 @@ public void stop() { // Make sure to entirely stop acceptor if any. CommonUtils.cancel(acceptWorker); - CommonUtils.join(acceptWorker, log); - CommonUtils.cancel(clientWorkers); - CommonUtils.join(clientWorkers, log); + + CommonUtils.join(acceptWorker, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); + CommonUtils.join(clientWorkers, CommonUtils.DFLT_WAIT_TO_STOP_TIMOEUT, log); filterChain.stop();