You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2023/07/04 04:07:50 UTC
[iotdb] branch master updated: Fix semantic incompatibility when AtomicReference replaces volatile (#10418)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a3c1b68d01e Fix semantic incompatibility when AtomicReference replaces volatile (#10418)
a3c1b68d01e is described below
commit a3c1b68d01ea0cb3b795fa941f890acf308d4201
Author: Potato <ta...@apache.org>
AuthorDate: Tue Jul 4 12:07:44 2023 +0800
Fix semantic incompatibility when AtomicReference replaces volatile (#10418)
Signed-off-by: OneSizeFitQuorum <ta...@apache.org>
---
.../org/apache/iotdb/confignode/manager/ConfigManager.java | 9 ++++-----
.../java/org/apache/iotdb/confignode/procedure/Procedure.java | 6 +++---
.../apache/iotdb/confignode/procedure/ProcedureExecutor.java | 10 +++++-----
.../java/org/apache/iotdb/consensus/ratis/RequestMessage.java | 7 +++----
.../java/org/apache/iotdb/consensus/ratis/ResponseMessage.java | 5 ++---
5 files changed, 17 insertions(+), 20 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
index b6ed07a869e..223c0372372 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java
@@ -208,7 +208,7 @@ public class ConfigManager implements IManager {
private static final CommonConfig COMMON_CONF = CommonDescriptor.getInstance().getConfig();
/** Manage PartitionTable read/write requests through the ConsensusLayer. */
- private AtomicReference<ConsensusManager> consensusManager;
+ private final AtomicReference<ConsensusManager> consensusManager = new AtomicReference<>();
/** Manage cluster node. */
private final NodeManager nodeManager;
@@ -305,12 +305,11 @@ public class ConfigManager implements IManager {
}
public void initConsensusManager() throws IOException {
- ConsensusManager consensusManager = new ConsensusManager(this, this.stateMachine);
- this.consensusManager = new AtomicReference<>(consensusManager);
+ this.consensusManager.set(new ConsensusManager(this, this.stateMachine));
}
public void close() throws IOException {
- if (consensusManager != null) {
+ if (consensusManager.get() != null) {
consensusManager.get().close();
}
if (partitionManager != null) {
@@ -1122,7 +1121,7 @@ public class ConfigManager implements IManager {
public TSStatus createPeerForConsensusGroup(List<TConfigNodeLocation> configNodeLocations) {
for (int i = 0; i < 30; i++) {
try {
- if (consensusManager == null) {
+ if (consensusManager.get() == null) {
Thread.sleep(1000);
} else {
// When add non Seed-ConfigNode to the ConfigNodeGroup, the parameter should be emptyList
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
index 36bc8afcb4c..d0b4a016fa1 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/Procedure.java
@@ -64,7 +64,7 @@ public abstract class Procedure<Env> implements Comparable<Procedure<Env>> {
private volatile long timeout = NO_TIMEOUT;
private volatile long lastUpdate;
- private AtomicReference<byte[]> result = null;
+ private final AtomicReference<byte[]> result = new AtomicReference<>();
private volatile boolean locked = false;
private boolean lockedWhenLoading = false;
@@ -173,7 +173,7 @@ public abstract class Procedure<Env> implements Comparable<Procedure<Env>> {
}
// result
- if (result != null) {
+ if (result.get() != null) {
stream.writeInt(result.get().length);
stream.write(result.get());
} else {
@@ -652,7 +652,7 @@ public abstract class Procedure<Env> implements Comparable<Procedure<Env>> {
* @param result the serialized result that will be passed to the client
*/
protected void setResult(byte[] result) {
- this.result = new AtomicReference<>(result);
+ this.result.set(result);
}
/**
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
index d8c5db84328..c988f210824 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java
@@ -710,7 +710,7 @@ public class ProcedureExecutor<Env> {
private class WorkerThread extends StoppableThread {
private final AtomicLong startTime = new AtomicLong(Long.MAX_VALUE);
- private AtomicReference<Procedure<Env>> activeProcedure;
+ private final AtomicReference<Procedure<Env>> activeProcedure = new AtomicReference<>();
protected long keepAliveTime = -1;
public WorkerThread(ThreadGroup threadGroup) {
@@ -736,19 +736,19 @@ public class ProcedureExecutor<Env> {
if (procedure == null) {
continue;
}
- this.activeProcedure = new AtomicReference<>(procedure);
+ this.activeProcedure.set(procedure);
int activeCount = activeExecutorCount.incrementAndGet();
startTime.set(System.currentTimeMillis());
executeProcedure(procedure);
activeCount = activeExecutorCount.decrementAndGet();
LOG.trace("Halt pid={}, activeCount={}", procedure.getProcId(), activeCount);
- this.activeProcedure = null;
+ this.activeProcedure.set(null);
lastUpdated = System.currentTimeMillis();
startTime.set(lastUpdated);
}
} catch (Throwable throwable) {
- if (this.activeProcedure != null) {
+ if (this.activeProcedure.get() != null) {
LOG.warn("Worker terminated {}", this.activeProcedure.get(), throwable);
}
} finally {
@@ -804,7 +804,7 @@ public class ProcedureExecutor<Env> {
// check if any of the worker is stuck
int stuckCount = 0;
for (WorkerThread worker : workerThreads) {
- if (worker.activeProcedure == null
+ if (worker.activeProcedure.get() == null
|| worker.getCurrentRunTime() < DEFAULT_WORKER_STUCK_THRESHOLD) {
continue;
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RequestMessage.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RequestMessage.java
index a150e5f123a..3c02ca42e2f 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RequestMessage.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RequestMessage.java
@@ -30,7 +30,7 @@ import java.util.concurrent.atomic.AtomicReference;
class RequestMessage implements Message {
private final IConsensusRequest actualRequest;
- private AtomicReference<ByteString> serializedContent = new AtomicReference<>();
+ private final AtomicReference<ByteString> serializedContent = new AtomicReference<>();
RequestMessage(IConsensusRequest request) {
this.actualRequest = request;
@@ -45,9 +45,8 @@ class RequestMessage implements Message {
if (serializedContent.get() == null) {
synchronized (this) {
if (serializedContent.get() == null) {
- serializedContent =
- new AtomicReference<>(
- UnsafeByteOperations.unsafeWrap(actualRequest.serializeToByteBuffer()));
+ serializedContent.set(
+ UnsafeByteOperations.unsafeWrap(actualRequest.serializeToByteBuffer()));
}
}
}
diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ResponseMessage.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ResponseMessage.java
index ca9cffa8962..ce1b7c14593 100644
--- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ResponseMessage.java
+++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ResponseMessage.java
@@ -38,7 +38,7 @@ class ResponseMessage implements Message {
*/
private final Object contentHolder;
- private AtomicReference<ByteString> serializedData = new AtomicReference<>();
+ private final AtomicReference<ByteString> serializedData = new AtomicReference<>();
private final Logger logger = LoggerFactory.getLogger(ResponseMessage.class);
ResponseMessage(Object content) {
@@ -57,8 +57,7 @@ class ResponseMessage implements Message {
assert contentHolder instanceof TSStatus;
TSStatus status = (TSStatus) contentHolder;
try {
- serializedData =
- new AtomicReference<>(ByteString.copyFrom(Utils.serializeTSStatus(status)));
+ serializedData.set(ByteString.copyFrom(Utils.serializeTSStatus(status)));
} catch (TException e) {
logger.warn("serialize TSStatus failed {}", status);
}