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 2021/07/28 02:27:40 UTC
[iotdb] branch delete_ping_client created (now 84cb973)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a change to branch delete_ping_client
in repository https://gitbox.apache.org/repos/asf/iotdb.git.
at 84cb973 fix
This branch includes the following new commits:
new 84cb973 fix
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: fix
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch delete_ping_client
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 84cb973abde11bb19d24d5dd4a46a9009e98fa0c
Author: LebronAl <TX...@gmail.com>
AuthorDate: Wed Jul 28 10:24:34 2021 +0800
fix
---
.../iotdb/cluster/client/DataClientProvider.java | 35 -----------
.../cluster/client/async/AsyncClientPool.java | 22 -------
.../iotdb/cluster/client/sync/SyncClientPool.java | 21 -------
.../cluster/server/member/MetaGroupMember.java | 71 ----------------------
4 files changed, 149 deletions(-)
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/client/DataClientProvider.java b/cluster/src/main/java/org/apache/iotdb/cluster/client/DataClientProvider.java
index 0950958..18a315d 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/client/DataClientProvider.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/client/DataClientProvider.java
@@ -76,22 +76,6 @@ public class DataClientProvider {
}
/**
- * Get a thrift client from the tail of deque that will connect to "node" using the data port for
- * refresh.
- *
- * @param node the node to be connected
- * @param timeout timeout threshold of connection
- */
- public AsyncDataClient getAsyncDataClientForRefresh(Node node, int timeout) throws IOException {
- AsyncDataClient client = (AsyncDataClient) getDataAsyncClientPool().getClientForRefresh(node);
- if (client == null) {
- throw new IOException(GET_CLIENT_FAILED_MSG + node);
- }
- client.setTimeout(timeout);
- return client;
- }
-
- /**
* IMPORTANT!!! After calling this function, the caller should make sure to call {@link
* org.apache.iotdb.cluster.utils.ClientUtils#putBackSyncClient(Client)} to put the client back
* into the client pool, otherwise there is a risk of client leakage.
@@ -109,23 +93,4 @@ public class DataClientProvider {
client.setTimeout(timeout);
return client;
}
-
- /**
- * IMPORTANT!!! After calling this function, the caller should make sure to call {@link
- * org.apache.iotdb.cluster.utils.ClientUtils#putBackSyncClient(Client)} to put the client back
- * into the client pool, otherwise there is a risk of client leakage.
- *
- * <p>Get a thrift client from the tail of deque that will connect to "node" using the data port.
- *
- * @param node the node to be connected
- * @param timeout timeout threshold of connection
- */
- public SyncDataClient getSyncDataClientForRefresh(Node node, int timeout) throws IOException {
- SyncDataClient client = (SyncDataClient) getDataSyncClientPool().getClientForRefresh(node);
- if (client == null) {
- throw new IOException(GET_CLIENT_FAILED_MSG + node);
- }
- client.setTimeout(timeout);
- return client;
- }
}
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java b/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java
index ddabff7..4ba3fe9 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/client/async/AsyncClientPool.java
@@ -53,28 +53,6 @@ public class AsyncClientPool {
}
/**
- * Get a client of the given node from the cache if one is available, or null.
- *
- * <p>IMPORTANT!!! The caller should check whether the return value is null or not!
- *
- * @param node the node want to connect
- * @return if the node can connect, return the client, otherwise null
- */
- public AsyncClient getClientForRefresh(Node node) {
- ClusterNode clusterNode = new ClusterNode(node);
- // As clientCaches is ConcurrentHashMap, computeIfAbsent is thread safety.
- Deque<AsyncClient> clientStack =
- clientCaches.computeIfAbsent(clusterNode, n -> new ArrayDeque<>());
- synchronized (clientStack) {
- if (clientStack.isEmpty()) {
- return null;
- } else {
- return clientStack.pollLast();
- }
- }
- }
-
- /**
* See getClient(Node node, boolean activatedOnly)
*
* @param node
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java b/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java
index f607fa3..2c279c0 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/client/sync/SyncClientPool.java
@@ -52,27 +52,6 @@ public class SyncClientPool {
}
/**
- * Get a client of the given node from the cache if one is available, or null.
- *
- * <p>IMPORTANT!!! The caller should check whether the return value is null or not!
- *
- * @param node the node want to connect
- * @return if the node can connect, return the client, otherwise null
- */
- public Client getClientForRefresh(Node node) {
- ClusterNode clusterNode = new ClusterNode(node);
- // As clientCaches is ConcurrentHashMap, computeIfAbsent is thread safety.
- Deque<Client> clientStack = clientCaches.computeIfAbsent(clusterNode, n -> new ArrayDeque<>());
- synchronized (clientStack) {
- if (clientStack.isEmpty()) {
- return null;
- } else {
- return clientStack.poll();
- }
- }
- }
-
- /**
* See getClient(Node node, boolean activatedOnly)
*
* @param node the node want to connect
diff --git a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java
index 1cc4ad9..60c54b2 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/member/MetaGroupMember.java
@@ -56,8 +56,6 @@ import org.apache.iotdb.cluster.rpc.thrift.CheckStatusResponse;
import org.apache.iotdb.cluster.rpc.thrift.HeartBeatRequest;
import org.apache.iotdb.cluster.rpc.thrift.HeartBeatResponse;
import org.apache.iotdb.cluster.rpc.thrift.Node;
-import org.apache.iotdb.cluster.rpc.thrift.RaftService;
-import org.apache.iotdb.cluster.rpc.thrift.RefreshReuqest;
import org.apache.iotdb.cluster.rpc.thrift.SendSnapshotRequest;
import org.apache.iotdb.cluster.rpc.thrift.StartUpStatus;
import org.apache.iotdb.cluster.rpc.thrift.TSMetaService;
@@ -165,12 +163,6 @@ public class MetaGroupMember extends RaftMember {
*/
private static final int REPORT_INTERVAL_SEC = 10;
- /**
- * every "REFRESH_CLIENT_SEC" seconds, a dataClientRefresher thread will try to refresh one thrift
- * connection for each nodes other than itself.
- */
- private static final int REFRESH_CLIENT_SEC = 1;
-
/** how many times is a data record replicated, also the number of nodes in a data group */
private static final int REPLICATION_NUM =
ClusterDescriptor.getInstance().getConfig().getReplicationNum();
@@ -219,8 +211,6 @@ public class MetaGroupMember extends RaftMember {
private DataClientProvider dataClientProvider;
- private ScheduledExecutorService dataClientRefresher;
-
/**
* a single thread pool, every "REPORT_INTERVAL_SEC" seconds, "reportThread" will print the status
* of all raft members in this node
@@ -341,8 +331,6 @@ public class MetaGroupMember extends RaftMember {
Executors.newSingleThreadScheduledExecutor(n -> new Thread(n, "NodeReportThread"));
hardLinkCleanerThread =
Executors.newSingleThreadScheduledExecutor(n -> new Thread(n, "HardLinkCleaner"));
- dataClientRefresher =
- Executors.newSingleThreadScheduledExecutor(n -> new Thread(n, "DataClientRefresher"));
}
/**
@@ -361,15 +349,6 @@ public class MetaGroupMember extends RaftMember {
if (clientServer != null) {
clientServer.stop();
}
- if (dataClientRefresher != null) {
- dataClientRefresher.shutdownNow();
- try {
- dataClientRefresher.awaitTermination(10, TimeUnit.SECONDS);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- logger.error("Unexpected interruption when waiting for reportThread to end", e);
- }
- }
if (reportThread != null) {
reportThread.shutdownNow();
try {
@@ -481,56 +460,6 @@ public class MetaGroupMember extends RaftMember {
CLEAN_HARDLINK_INTERVAL_SEC,
CLEAN_HARDLINK_INTERVAL_SEC,
TimeUnit.SECONDS);
- dataClientRefresher.scheduleAtFixedRate(
- this::refreshClientOnce, REFRESH_CLIENT_SEC, REFRESH_CLIENT_SEC, TimeUnit.SECONDS);
- }
-
- private void refreshClientOnce() {
- for (Node receiver : allNodes) {
- if (!receiver.equals(thisNode)) {
- if (ClusterDescriptor.getInstance().getConfig().isUseAsyncServer()) {
- refreshClientOnceAsync(receiver);
- } else {
- refreshClientOnceSync(receiver);
- }
- }
- }
- }
-
- private void refreshClientOnceSync(Node receiver) {
- RaftService.Client client = null;
- try {
- client =
- getClientProvider()
- .getSyncDataClientForRefresh(receiver, RaftServer.getWriteOperationTimeoutMS());
- RefreshReuqest req = new RefreshReuqest();
- client.refreshConnection(req);
- } catch (IOException ignored) {
- } catch (TException e) {
- logger.info("encounter refreshing client timeout, throw broken connection", e);
- // the connection may be broken, close it to avoid it being reused
- client.getInputProtocol().getTransport().close();
- } finally {
- if (client != null) {
- ClientUtils.putBackSyncClient(client);
- }
- }
- }
-
- private void refreshClientOnceAsync(Node receiver) {
- RaftService.AsyncClient client;
- try {
- client =
- getClientProvider()
- .getAsyncDataClientForRefresh(receiver, RaftServer.getWriteOperationTimeoutMS());
- } catch (IOException e) {
- return;
- }
- try {
- client.refreshConnection(new RefreshReuqest(), new GenericHandler<>(receiver, null));
- } catch (TException e) {
- logger.info("encounter refreshing client timeout, throw broken connection", e);
- }
}
private void generateNodeReport() {