sashapolo commented on code in PR #3068: URL: https://github.com/apache/ignite-3/pull/3068#discussion_r1461512048
########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorker.java: ########## @@ -0,0 +1,41 @@ +/* + * 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.worker; + +/** + * A worker that performs some critical work and hence need to be monitored for being hung. Review Comment: ```suggestion * A worker that performs some critical work and hence needs to be monitored for being blocked. ``` ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalSingleThreadExecutor.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.worker; + +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Single thread executor instrumented to be used as a {@link CriticalWorker} and being monitorable by the {@link CriticalWorkerWatchdog}. + * Must be registered with the watchdog explicitly. + */ +public class CriticalSingleThreadExecutor extends ThreadPoolExecutor implements CriticalWorker { + + private Thread lastSeenThread; + private volatile long heartbeatNanos = NOT_MONITORED; + + public CriticalSingleThreadExecutor(ThreadFactory threadFactory) { + super(1, 1, 0, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), threadFactory); + } + + @Override + protected void beforeExecute(Thread t, Runnable r) { + try { Review Comment: Why do we need try-finally here? ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.worker; + +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.jetbrains.annotations.Nullable; + +/** + * A watchdog that monitors liveness of the registered workers and, if a worker is found to be hung, logs the corresopnding + * information (including the stack trace corresponding to the woeker's thread) and then calls the failure handler + * to fail the node. + * + * <p>Each worker is expected to maintain its {@link CriticalWorker#heartbeatNanos()} growing while the worker executes some computations. + * If the worker does not do any computations (it is blocked on an I/O operation, waits for a lock, or has no work to do), + * it must set its {@link CriticalWorker#heartbeatNanos()} to {@link CriticalWorker#NOT_MONITORED}. + * + * <p>The watchdog periodically does its check; if it finds a worker that lags more than allowed and it is not in Review Comment: ```suggestion * <p>The watchdog periodically performs a check; if it finds a worker that lags more than allowed and it is not in the ``` ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalSingleThreadExecutor.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.worker; + +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Single thread executor instrumented to be used as a {@link CriticalWorker} and being monitorable by the {@link CriticalWorkerWatchdog}. Review Comment: ```suggestion * Single thread executor instrumented to be used as a {@link CriticalWorker} and being monitored by the {@link CriticalWorkerWatchdog}. ``` ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.worker; + +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.jetbrains.annotations.Nullable; + +/** + * A watchdog that monitors liveness of the registered workers and, if a worker is found to be hung, logs the corresopnding + * information (including the stack trace corresponding to the woeker's thread) and then calls the failure handler Review Comment: What failure handler does it call? ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorker.java: ########## @@ -0,0 +1,41 @@ +/* + * 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.worker; + +/** + * A worker that performs some critical work and hence need to be monitored for being hung. + */ +public interface CriticalWorker { + /** + * Hearbeat timestamp 'value' corresponding to the state in which the thread is not running some computation during Review Comment: ```suggestion * Heartbeat timestamp 'value' corresponding to the state in which the thread is not running any computations during ``` ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.worker; + +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.jetbrains.annotations.Nullable; + +/** + * A watchdog that monitors liveness of the registered workers and, if a worker is found to be hung, logs the corresopnding + * information (including the stack trace corresponding to the woeker's thread) and then calls the failure handler Review Comment: ```suggestion * information (including the stack trace corresponding to the worker's thread) and then calls the failure handler ``` ########## modules/network/src/main/java/org/apache/ignite/network/DefaultMessagingService.java: ########## @@ -110,18 +115,20 @@ public DefaultMessagingService( TopologyService topologyService, StaleIdDetector staleIdDetector, ClassDescriptorRegistry classDescriptorRegistry, - UserObjectMarshaller marshaller + UserObjectMarshaller marshaller, + CriticalWorkerRegistry criticalWorkerRegistry ) { this.factory = factory; this.topologyService = topologyService; this.staleIdDetector = staleIdDetector; this.classDescriptorRegistry = classDescriptorRegistry; this.marshaller = marshaller; + this.criticalWorkerRegistry = criticalWorkerRegistry; - this.outboundExecutor = Executors.newSingleThreadExecutor(NamedThreadFactory.create(nodeName, "MessagingService-outbound-", LOG)); + this.outboundExecutor = new CriticalSingleThreadExecutor(NamedThreadFactory.create(nodeName, "MessagingService-outbound", LOG)); Review Comment: Why did you remove the "-" at the end? ########## modules/network/src/main/java/org/apache/ignite/network/NettyWorkersRegistrar.java: ########## @@ -0,0 +1,147 @@ +/* + * 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.network; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.util.concurrent.EventExecutor; +import io.netty.util.concurrent.SingleThreadEventExecutor; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.apache.ignite.internal.worker.CriticalWorker; +import org.apache.ignite.internal.worker.CriticalWorkerRegistry; +import org.jetbrains.annotations.Nullable; + +/** + * This component is responsible for registering Netty workers with the {@link CriticalWorkerRegistry} and for updating their heartbeats. + */ +public class NettyWorkersRegistrar implements IgniteComponent { + /* + * It seems impossible to instrument tasks executed by Netty event loops, so the strategy we use is to + * send 'heartbeat' tasks to each of the event loops periodically; this tasks just update the heartbeat timestamp + * of the worker corresponding to an event loop. If an event loop's thread is hung, the hearbeat will not Review Comment: ```suggestion * of the worker corresponding to an event loop. If an event loop's thread is blocked, the hearbeat will not ``` ########## modules/network/src/main/java/org/apache/ignite/network/NettyWorkersRegistrar.java: ########## @@ -0,0 +1,147 @@ +/* + * 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.network; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.util.concurrent.EventExecutor; +import io.netty.util.concurrent.SingleThreadEventExecutor; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.apache.ignite.internal.worker.CriticalWorker; +import org.apache.ignite.internal.worker.CriticalWorkerRegistry; +import org.jetbrains.annotations.Nullable; + +/** + * This component is responsible for registering Netty workers with the {@link CriticalWorkerRegistry} and for updating their heartbeats. + */ +public class NettyWorkersRegistrar implements IgniteComponent { + /* + * It seems impossible to instrument tasks executed by Netty event loops, so the strategy we use is to + * send 'heartbeat' tasks to each of the event loops periodically; this tasks just update the heartbeat timestamp + * of the worker corresponding to an event loop. If an event loop's thread is hung, the hearbeat will not + * be updated, and the worker watchdog will treat the event loop worker as blocked. + */ + + private final CriticalWorkerRegistry criticalWorkerRegistry; + + private final ScheduledExecutorService scheduler; + + private final NettyBootstrapFactory bootstrapFactory; + + private final List<NettyWorker> workers = new CopyOnWriteArrayList<>(); Review Comment: Looks like this thing can simply be a volatile immutable list ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.worker; + +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.jetbrains.annotations.Nullable; + +/** + * A watchdog that monitors liveness of the registered workers and, if a worker is found to be hung, logs the corresopnding + * information (including the stack trace corresponding to the woeker's thread) and then calls the failure handler + * to fail the node. + * + * <p>Each worker is expected to maintain its {@link CriticalWorker#heartbeatNanos()} growing while the worker executes some computations. + * If the worker does not do any computations (it is blocked on an I/O operation, waits for a lock, or has no work to do), + * it must set its {@link CriticalWorker#heartbeatNanos()} to {@link CriticalWorker#NOT_MONITORED}. + * + * <p>The watchdog periodically does its check; if it finds a worker that lags more than allowed and it is not in + * NOT_MONITORED state, then logging and failure handling notification is triggered. + */ +public class CriticalWorkerWatchdog implements CriticalWorkerRegistry, IgniteComponent { + private final IgniteLogger log = Loggers.forClass(CriticalWorkerWatchdog.class); + + // TODO: IGNITE-21227 - make this configurable. + private static final long LIVENESS_CHECK_INTERVAL_MS = 200; + private static final long MAX_ALLOWED_LAG_MS = 500; + + private final ScheduledExecutorService scheduler; + + private final Set<CriticalWorker> registeredWorkers = ConcurrentHashMap.newKeySet(); + + @Nullable + private volatile ScheduledFuture<?> livenessProbeTaskFuture; + + public CriticalWorkerWatchdog(ScheduledExecutorService scheduler) { + this.scheduler = scheduler; + } + + @Override + public void register(CriticalWorker worker) { + registeredWorkers.add(worker); + } + + @Override + public void unregister(CriticalWorker worker) { + registeredWorkers.remove(worker); + } + + @Override + public CompletableFuture<Void> start() { + livenessProbeTaskFuture = scheduler.scheduleAtFixedRate( + this::probeLiveness, + LIVENESS_CHECK_INTERVAL_MS, + LIVENESS_CHECK_INTERVAL_MS, + TimeUnit.MILLISECONDS + ); + + return nullCompletedFuture(); + } + + private void probeLiveness() { + long nowNanos = System.nanoTime(); + + for (CriticalWorker worker : registeredWorkers) { + long heartbeatNanos = worker.heartbeatNanos(); + + if (heartbeatNanos == CriticalWorker.NOT_MONITORED) { + continue; + } + + long delayMillis = TimeUnit.NANOSECONDS.toMillis(nowNanos - heartbeatNanos); + if (delayMillis > MAX_ALLOWED_LAG_MS) { + ThreadMXBean bean = ManagementFactory.getThreadMXBean(); + + ThreadInfo[] threadInfos = bean.getThreadInfo(new long[]{worker.threadId()}, true, true, Integer.MAX_VALUE); Review Comment: Should we collect all blocked thread IDs first and only then call `getThreadInfo`? ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalSingleThreadExecutor.java: ########## @@ -0,0 +1,70 @@ +/* + * 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.worker; + +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +/** + * Single thread executor instrumented to be used as a {@link CriticalWorker} and being monitorable by the {@link CriticalWorkerWatchdog}. + * Must be registered with the watchdog explicitly. + */ +public class CriticalSingleThreadExecutor extends ThreadPoolExecutor implements CriticalWorker { + + private Thread lastSeenThread; Review Comment: Are you sure that this field shouldn't be volatile? ########## modules/network/src/main/java/org/apache/ignite/network/NettyWorkersRegistrar.java: ########## @@ -0,0 +1,147 @@ +/* + * 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.network; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.util.concurrent.EventExecutor; +import io.netty.util.concurrent.SingleThreadEventExecutor; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.apache.ignite.internal.worker.CriticalWorker; +import org.apache.ignite.internal.worker.CriticalWorkerRegistry; +import org.jetbrains.annotations.Nullable; + +/** + * This component is responsible for registering Netty workers with the {@link CriticalWorkerRegistry} and for updating their heartbeats. + */ +public class NettyWorkersRegistrar implements IgniteComponent { + /* + * It seems impossible to instrument tasks executed by Netty event loops, so the strategy we use is to + * send 'heartbeat' tasks to each of the event loops periodically; this tasks just update the heartbeat timestamp + * of the worker corresponding to an event loop. If an event loop's thread is hung, the hearbeat will not + * be updated, and the worker watchdog will treat the event loop worker as blocked. + */ + + private final CriticalWorkerRegistry criticalWorkerRegistry; + + private final ScheduledExecutorService scheduler; + + private final NettyBootstrapFactory bootstrapFactory; + + private final List<NettyWorker> workers = new CopyOnWriteArrayList<>(); + + @Nullable + private volatile ScheduledFuture<?> sendHearbeatsTaskFuture; + + /** + * Constructor. + * + * @param criticalWorkerRegistry Used to register workers representing Netty event loops to detect when their threads are + * blocked. + * @param scheduler Used to schedule periodic tasks. + * @param bootstrapFactory Used to obtain Netty workers. + */ + public NettyWorkersRegistrar( + CriticalWorkerRegistry criticalWorkerRegistry, + ScheduledExecutorService scheduler, + NettyBootstrapFactory bootstrapFactory + ) { + this.criticalWorkerRegistry = criticalWorkerRegistry; + this.scheduler = scheduler; + this.bootstrapFactory = bootstrapFactory; + } + + @Override + public CompletableFuture<Void> start() { + for (NioEventLoopGroup group : bootstrapFactory.eventLoopGroups()) { + registerWorkersFor(group); + } + + sendHearbeatsTaskFuture = scheduler.scheduleAtFixedRate(this::sendHearbeats, 100, 100, MILLISECONDS); Review Comment: Should we introduce some constants? ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.worker; + +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.jetbrains.annotations.Nullable; + +/** + * A watchdog that monitors liveness of the registered workers and, if a worker is found to be hung, logs the corresopnding Review Comment: ```suggestion * A watchdog that monitors liveness of the registered workers and, if a worker is found to be blocked, logs the corresponding ``` ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.worker; + +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.jetbrains.annotations.Nullable; + +/** + * A watchdog that monitors liveness of the registered workers and, if a worker is found to be hung, logs the corresopnding + * information (including the stack trace corresponding to the woeker's thread) and then calls the failure handler + * to fail the node. + * + * <p>Each worker is expected to maintain its {@link CriticalWorker#heartbeatNanos()} growing while the worker executes some computations. + * If the worker does not do any computations (it is blocked on an I/O operation, waits for a lock, or has no work to do), + * it must set its {@link CriticalWorker#heartbeatNanos()} to {@link CriticalWorker#NOT_MONITORED}. + * + * <p>The watchdog periodically does its check; if it finds a worker that lags more than allowed and it is not in + * NOT_MONITORED state, then logging and failure handling notification is triggered. Review Comment: ```suggestion * NOT_MONITORED state, then a logging and a failure handling notification is triggered. ``` ########## modules/network/src/main/java/org/apache/ignite/network/DefaultMessagingService.java: ########## @@ -479,6 +486,13 @@ private boolean isSelf(@Nullable String consistentId, InetSocketAddress targetAd return false; } + /** + * STarts the service. Review Comment: ```suggestion * Starts the service. ``` ########## modules/core/src/testFixtures/java/org/apache/ignite/internal/worker/NullCriticalWorkerRegistry.java: ########## @@ -0,0 +1,33 @@ +/* + * 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.worker; + +/** + * {@link CriticalWorkerRegistry} implementation that does nothing. + */ +public class NullCriticalWorkerRegistry implements CriticalWorkerRegistry { Review Comment: I would suggest to rename this class to `NoOpCriticalWorkerRegistry` ########## modules/network/src/main/java/org/apache/ignite/network/NettyWorkersRegistrar.java: ########## @@ -0,0 +1,147 @@ +/* + * 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.network; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.util.concurrent.EventExecutor; +import io.netty.util.concurrent.SingleThreadEventExecutor; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.apache.ignite.internal.worker.CriticalWorker; +import org.apache.ignite.internal.worker.CriticalWorkerRegistry; +import org.jetbrains.annotations.Nullable; + +/** + * This component is responsible for registering Netty workers with the {@link CriticalWorkerRegistry} and for updating their heartbeats. + */ +public class NettyWorkersRegistrar implements IgniteComponent { + /* + * It seems impossible to instrument tasks executed by Netty event loops, so the strategy we use is to + * send 'heartbeat' tasks to each of the event loops periodically; this tasks just update the heartbeat timestamp Review Comment: ```suggestion * send 'heartbeat' tasks to each of the event loops periodically; these tasks just update the heartbeat timestamp ``` ########## modules/network/src/main/java/org/apache/ignite/network/DefaultMessagingService.java: ########## @@ -531,4 +547,36 @@ public void stopDroppingMessages() { public ConnectionManager connectionManager() { return connectionManager; } + + private static class CriticalLazyStripedExecutor extends LazyStripedExecutor { + private final CriticalWorkerRegistry workerRegistry; + + private final List<CriticalWorker> registeredWorkers = new CopyOnWriteArrayList<>(); + + CriticalLazyStripedExecutor(String nodeName, String poolName, CriticalWorkerRegistry workerRegistry) { + super(nodeName, poolName); + + this.workerRegistry = workerRegistry; + } + + @Override + protected ExecutorService newSingleThreadExecutor(NamedThreadFactory threadFactory) { + CriticalSingleThreadExecutor executor = new CriticalSingleThreadExecutor(threadFactory); + + workerRegistry.register(executor); + + registeredWorkers.add(executor); + + return executor; + } + + @Override + protected void onStoppingInitiated() { + super.onStoppingInitiated(); + + for (CriticalWorker worker : registeredWorkers) { + workerRegistry.unregister(worker); Review Comment: Maybe we should have a method like `close` or `unregisterAll` on the registry instead? ########## modules/core/src/main/java/org/apache/ignite/internal/worker/CriticalWorkerWatchdog.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.worker; + +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.jetbrains.annotations.Nullable; + +/** + * A watchdog that monitors liveness of the registered workers and, if a worker is found to be hung, logs the corresopnding + * information (including the stack trace corresponding to the woeker's thread) and then calls the failure handler + * to fail the node. + * + * <p>Each worker is expected to maintain its {@link CriticalWorker#heartbeatNanos()} growing while the worker executes some computations. + * If the worker does not do any computations (it is blocked on an I/O operation, waits for a lock, or has no work to do), + * it must set its {@link CriticalWorker#heartbeatNanos()} to {@link CriticalWorker#NOT_MONITORED}. + * + * <p>The watchdog periodically does its check; if it finds a worker that lags more than allowed and it is not in + * NOT_MONITORED state, then logging and failure handling notification is triggered. + */ +public class CriticalWorkerWatchdog implements CriticalWorkerRegistry, IgniteComponent { + private final IgniteLogger log = Loggers.forClass(CriticalWorkerWatchdog.class); + + // TODO: IGNITE-21227 - make this configurable. + private static final long LIVENESS_CHECK_INTERVAL_MS = 200; + private static final long MAX_ALLOWED_LAG_MS = 500; + + private final ScheduledExecutorService scheduler; + + private final Set<CriticalWorker> registeredWorkers = ConcurrentHashMap.newKeySet(); + + @Nullable + private volatile ScheduledFuture<?> livenessProbeTaskFuture; + + public CriticalWorkerWatchdog(ScheduledExecutorService scheduler) { + this.scheduler = scheduler; + } + + @Override + public void register(CriticalWorker worker) { + registeredWorkers.add(worker); + } + + @Override + public void unregister(CriticalWorker worker) { + registeredWorkers.remove(worker); + } + + @Override + public CompletableFuture<Void> start() { + livenessProbeTaskFuture = scheduler.scheduleAtFixedRate( + this::probeLiveness, + LIVENESS_CHECK_INTERVAL_MS, + LIVENESS_CHECK_INTERVAL_MS, + TimeUnit.MILLISECONDS + ); + + return nullCompletedFuture(); + } + + private void probeLiveness() { + long nowNanos = System.nanoTime(); + + for (CriticalWorker worker : registeredWorkers) { + long heartbeatNanos = worker.heartbeatNanos(); + + if (heartbeatNanos == CriticalWorker.NOT_MONITORED) { + continue; + } + + long delayMillis = TimeUnit.NANOSECONDS.toMillis(nowNanos - heartbeatNanos); + if (delayMillis > MAX_ALLOWED_LAG_MS) { + ThreadMXBean bean = ManagementFactory.getThreadMXBean(); + + ThreadInfo[] threadInfos = bean.getThreadInfo(new long[]{worker.threadId()}, true, true, Integer.MAX_VALUE); Review Comment: ```suggestion ThreadInfo[] threadInfos = bean.getThreadInfo(new long[]{worker.threadId()}, true, true); ``` ########## modules/network/src/main/java/org/apache/ignite/network/NettyWorkersRegistrar.java: ########## @@ -0,0 +1,147 @@ +/* + * 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.network; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture; + +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.util.concurrent.EventExecutor; +import io.netty.util.concurrent.SingleThreadEventExecutor; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import org.apache.ignite.internal.manager.IgniteComponent; +import org.apache.ignite.internal.worker.CriticalWorker; +import org.apache.ignite.internal.worker.CriticalWorkerRegistry; +import org.jetbrains.annotations.Nullable; + +/** + * This component is responsible for registering Netty workers with the {@link CriticalWorkerRegistry} and for updating their heartbeats. + */ +public class NettyWorkersRegistrar implements IgniteComponent { + /* + * It seems impossible to instrument tasks executed by Netty event loops, so the strategy we use is to + * send 'heartbeat' tasks to each of the event loops periodically; this tasks just update the heartbeat timestamp + * of the worker corresponding to an event loop. If an event loop's thread is hung, the hearbeat will not + * be updated, and the worker watchdog will treat the event loop worker as blocked. + */ + + private final CriticalWorkerRegistry criticalWorkerRegistry; + + private final ScheduledExecutorService scheduler; + + private final NettyBootstrapFactory bootstrapFactory; + + private final List<NettyWorker> workers = new CopyOnWriteArrayList<>(); + + @Nullable + private volatile ScheduledFuture<?> sendHearbeatsTaskFuture; + + /** + * Constructor. + * + * @param criticalWorkerRegistry Used to register workers representing Netty event loops to detect when their threads are + * blocked. + * @param scheduler Used to schedule periodic tasks. + * @param bootstrapFactory Used to obtain Netty workers. + */ + public NettyWorkersRegistrar( + CriticalWorkerRegistry criticalWorkerRegistry, + ScheduledExecutorService scheduler, + NettyBootstrapFactory bootstrapFactory + ) { + this.criticalWorkerRegistry = criticalWorkerRegistry; + this.scheduler = scheduler; + this.bootstrapFactory = bootstrapFactory; + } + + @Override + public CompletableFuture<Void> start() { + for (NioEventLoopGroup group : bootstrapFactory.eventLoopGroups()) { + registerWorkersFor(group); + } + + sendHearbeatsTaskFuture = scheduler.scheduleAtFixedRate(this::sendHearbeats, 100, 100, MILLISECONDS); + + return nullCompletedFuture(); + } + + private void registerWorkersFor(NioEventLoopGroup group) { + List<NettyWorker> groupWorkers = new ArrayList<>(group.executorCount()); + + for (EventExecutor eventExecutor : group) { + SingleThreadEventExecutor executor = (SingleThreadEventExecutor) eventExecutor; + groupWorkers.add(new NettyWorker(executor)); + } + + for (NettyWorker worker : groupWorkers) { + criticalWorkerRegistry.register(worker); + } + + workers.addAll(groupWorkers); + } + + private void sendHearbeats() { + for (NettyWorker worker : workers) { + worker.sendHeartbeat(); + } + } + + @Override + public void stop() throws Exception { + Future<?> heartBeatsTaskFuture = sendHearbeatsTaskFuture; + if (heartBeatsTaskFuture != null) { + heartBeatsTaskFuture.cancel(false); + } + + for (NettyWorker worker : workers) { + criticalWorkerRegistry.unregister(worker); + } + } + + private static class NettyWorker implements CriticalWorker { + private final SingleThreadEventExecutor eventExecutor; + + private volatile long heartbeatNanos = System.nanoTime(); + + private final Runnable sendHeartbeatTask = () -> heartbeatNanos = System.nanoTime(); Review Comment: Can it simply be a method? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
