Skip to content

Commit a4c9f4e

Browse files
lvs007lvs007
committed
add tcp flow controller
1 parent 161aeb9 commit a4c9f4e

7 files changed

Lines changed: 147 additions & 24 deletions

File tree

‎src/main/java/org/tron/common/overlay/discover/node/statistics/MessageCountStatistics.java‎

Lines changed: 27 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -7,23 +7,23 @@ public class MessageCountStatistics {
77

88
private static int SIZE = 60;
99

10-
private int [] szCount = new int[SIZE];
10+
private int[] szCount = new int[SIZE];
1111

1212
private long indexTime = System.currentTimeMillis() / 1000;
1313

14-
private int index = (int)(indexTime % SIZE);
14+
private int index = (int) (indexTime % SIZE);
1515

1616
private long totalCount = 0;
1717

18-
private void update(){
19-
long time = System.currentTimeMillis() / 1000;
20-
long gap = time - indexTime;
21-
int k = gap > SIZE ? SIZE : (int)gap;
22-
if (k > 0){
23-
for (int i = 1; i <= k; i++){
18+
private void update() {
19+
long time = System.currentTimeMillis() / 1000;
20+
long gap = time - indexTime;
21+
int k = gap > SIZE ? SIZE : (int) gap;
22+
if (k > 0) {
23+
for (int i = 1; i <= k; i++) {
2424
szCount[(index + i) % SIZE] = 0;
2525
}
26-
index = (int)(time % SIZE);
26+
index = (int) (time % SIZE);
2727
indexTime = time;
2828
}
2929
}
@@ -34,14 +34,20 @@ public void add() {
3434
totalCount++;
3535
}
3636

37-
public int getCount(int interval){
38-
if (interval > SIZE){
37+
public void add(int length) {
38+
update();
39+
szCount[index]++;
40+
totalCount += length;
41+
}
42+
43+
public int getCount(int interval) {
44+
if (interval > SIZE) {
3945
logger.warn("Param interval({}) is gt SIZE({})", interval, SIZE);
4046
return 0;
4147
}
4248
update();
4349
int count = 0;
44-
for (int i = 0; i < interval; i++){
50+
for (int i = 0; i < interval; i++) {
4551
count += szCount[(SIZE + index - i) % SIZE];
4652
}
4753
return count;
@@ -50,4 +56,13 @@ public int getCount(int interval){
5056
public long getTotalCount() {
5157
return totalCount;
5258
}
59+
60+
public void reset() {
61+
totalCount = 0;
62+
}
63+
64+
@Override
65+
public String toString() {
66+
return String.valueOf(totalCount);
67+
}
5368
}

‎src/main/java/org/tron/common/overlay/discover/node/statistics/NodeStatistics.java‎

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,7 @@
2020

2121
import static java.lang.Math.min;
2222

23-
import java.util.Date;
2423
import java.util.concurrent.atomic.AtomicLong;
25-
import org.joda.time.DateTime;
2624
import org.tron.common.overlay.discover.node.Node;
2725
import org.tron.protos.Protocol.ReasonCode;
2826

@@ -31,6 +29,7 @@ public class NodeStatistics {
3129
public final static int REPUTATION_PREDEFINED = 100000;
3230
public final static long TOO_MANY_PEERS_PENALIZE_TIMEOUT = 60 * 1000L;
3331
private static final long CLEAR_CYCLE_TIME = 60 * 60 * 1000L;
32+
private static final long MIN_DATA_LENGTH = 2048L;
3433

3534
private boolean isPredefined = false;
3635

@@ -64,6 +63,9 @@ public class NodeStatistics {
6463
private long lastDisconnectedTime = 0;
6564
private long firstDisconnectedTime = 0;
6665

66+
//tcp flow stat
67+
public final MessageCountStatistics tcpFlow = new MessageCountStatistics();
68+
6769

6870
public NodeStatistics(Node node) {
6971
discoverMessageLatency = new SimpleStatter(node.getIdString());
@@ -203,6 +205,10 @@ public void setPredefined(boolean isPredefined) {
203205
this.isPredefined = isPredefined;
204206
}
205207

208+
public boolean isPredefined() {
209+
return isPredefined;
210+
}
211+
206212
public void setPersistedReputation(int persistedReputation) {
207213
this.persistedReputation = persistedReputation;
208214
}
@@ -219,7 +225,8 @@ public String toString() {
219225
", tron: " + tronInMessage + "/" + tronOutMessage + " " +
220226
(wasDisconnected() ? "X " + disconnectTimes : "") +
221227
(tronLastLocalDisconnectReason != null ? ("<=" + tronLastLocalDisconnectReason) : " ") +
222-
(tronLastRemoteDisconnectReason != null ? ("=>" + tronLastRemoteDisconnectReason) : " ");
228+
(tronLastRemoteDisconnectReason != null ? ("=>" + tronLastRemoteDisconnectReason) : " ") +
229+
", tcp flow: " + tcpFlow.getTotalCount();
223230
}
224231

225232
public class SimpleStatter {
@@ -261,4 +268,12 @@ public String getName() {
261268

262269
}
263270

271+
public boolean nodeIsHaveDataTransfer() {
272+
return tcpFlow.getTotalCount() > MIN_DATA_LENGTH;
273+
}
274+
275+
public void resetTcpFlow() {
276+
tcpFlow.reset();
277+
}
278+
264279
}

‎src/main/java/org/tron/common/overlay/message/MessageCodec.java‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,14 +3,7 @@
33
import io.netty.buffer.ByteBuf;
44
import io.netty.channel.ChannelHandlerContext;
55
import io.netty.handler.codec.ByteToMessageDecoder;
6-
import java.io.IOException;
76
import java.util.List;
8-
import org.apache.commons.lang3.ArrayUtils;
9-
import org.slf4j.Logger;
10-
import org.slf4j.LoggerFactory;
11-
import org.spongycastle.util.encoders.Hex;
12-
import org.springframework.beans.factory.annotation.Autowired;
13-
import org.springframework.context.ApplicationContext;
147
import org.springframework.context.annotation.Scope;
158
import org.springframework.stereotype.Component;
169
import org.tron.common.overlay.server.Channel;
@@ -28,11 +21,13 @@ public class MessageCodec extends ByteToMessageDecoder {
2821

2922
@Override
3023
protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
31-
byte[] encoded = new byte[buffer.readableBytes()];
24+
int length = buffer.readableBytes();
25+
byte[] encoded = new byte[length];
3226
buffer.readBytes(encoded);
3327
try {
3428
Message msg = createMessage(encoded);
3529
channel.getNodeStatistics().tronInMessage.add();
30+
channel.getNodeStatistics().tcpFlow.add(length);
3631
out.add(msg);
3732
} catch (Exception e) {
3833
channel.processException(e);

‎src/main/java/org/tron/common/overlay/server/Channel.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,8 @@ public void init(ChannelPipeline pipeline, String remoteId, boolean discoveryMod
107107

108108
isActive = remoteId != null && !remoteId.isEmpty();
109109

110+
startTime = System.currentTimeMillis();
111+
110112
//TODO: use config here
111113
pipeline.addLast("readTimeoutHandler", new ReadTimeoutHandler(60, TimeUnit.SECONDS));
112114
pipeline.addLast(stats.tcp);

‎src/main/java/org/tron/common/overlay/server/PeerConnectionCheckService.java‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
package org.tron.common.overlay.server;
2+
3+
import java.util.ArrayList;
4+
import java.util.Collections;
5+
import java.util.List;
6+
import java.util.concurrent.Executors;
7+
import java.util.concurrent.ScheduledExecutorService;
8+
import java.util.concurrent.TimeUnit;
9+
import javax.annotation.PostConstruct;
10+
import javax.annotation.PreDestroy;
11+
import lombok.extern.slf4j.Slf4j;
12+
import org.springframework.beans.factory.annotation.Autowired;
13+
import org.springframework.stereotype.Service;
14+
import org.tron.common.overlay.discover.node.statistics.NodeStatistics;
15+
import org.tron.core.config.args.Args;
16+
import org.tron.core.net.peer.PeerConnection;
17+
import org.tron.protos.Protocol.ReasonCode;
18+
19+
@Slf4j
20+
@Service
21+
public class PeerConnectionCheckService {
22+
23+
public static final long CHECK_TIME = 5 * 60 * 1000L;
24+
private double disconnectNumberFactor = Args.getInstance().getDisconnectNumberFactor();
25+
private double maxConnectNumberFactor = Args.getInstance().getMaxConnectNumberFactor();
26+
27+
@Autowired
28+
private SyncPool pool;
29+
30+
@Autowired
31+
private ChannelManager channelManager;
32+
33+
private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1,
34+
r -> new Thread(r, "check-peer-connect"));
35+
36+
@PostConstruct
37+
public void check() {
38+
logger.info("start the PeerConnectionCheckService");
39+
scheduledExecutorService
40+
.scheduleWithFixedDelay(new CheckDataTransferTask(), 5, 5, TimeUnit.MINUTES);
41+
}
42+
43+
@PreDestroy
44+
public void destroy() {
45+
scheduledExecutorService.shutdown();
46+
}
47+
48+
private class CheckDataTransferTask implements Runnable {
49+
50+
@Override
51+
public void run() {
52+
List<PeerConnection> peerConnectionList = pool.getActivePeers();
53+
List<Channel> willDisconnectPeerList = new ArrayList<>();
54+
for (PeerConnection peerConnection : peerConnectionList) {
55+
NodeStatistics nodeStatistics = peerConnection.getNodeStatistics();
56+
if (!nodeStatistics.nodeIsHaveDataTransfer()
57+
&& System.currentTimeMillis() - peerConnection.getStartTime() >= CHECK_TIME
58+
&& !channelManager.getTrustPeers().containsKey(peerConnection.getInetAddress())
59+
&& !nodeStatistics.isPredefined()) {
60+
//&& !peerConnection.isActive()
61+
//if xxx minutes not have data transfer,disconnect the peer,exclude trust peer and active peer
62+
willDisconnectPeerList.add(peerConnection);
63+
}
64+
nodeStatistics.resetTcpFlow();
65+
}
66+
if (!willDisconnectPeerList.isEmpty() && peerConnectionList.size()
67+
> Args.getInstance().getNodeMaxActiveNodes() * maxConnectNumberFactor) {
68+
Collections.shuffle(willDisconnectPeerList);
69+
for (int i = 0; i < willDisconnectPeerList.size() * disconnectNumberFactor; i++) {
70+
logger.error("{} not have data transfer, disconnect the peer",
71+
willDisconnectPeerList.get(i).getInetAddress());
72+
willDisconnectPeerList.get(i).disconnect(ReasonCode.TOO_MANY_PEERS);
73+
}
74+
}
75+
}
76+
}
77+
78+
}

‎src/main/java/org/tron/core/config/args/Args.java‎

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@
3434
import org.tron.common.crypto.ECKey;
3535
import org.tron.common.overlay.discover.node.Node;
3636
import org.tron.common.utils.ByteArray;
37-
import org.tron.common.utils.StringUtil;
3837
import org.tron.core.Constant;
3938
import org.tron.core.Wallet;
4039
import org.tron.core.config.Configuration;
@@ -273,6 +272,14 @@ public class Args {
273272
@Setter
274273
private double activeConnectFactor;
275274

275+
@Getter
276+
@Setter
277+
private double disconnectNumberFactor;
278+
279+
@Getter
280+
@Setter
281+
private double maxConnectNumberFactor;
282+
276283
public static void clearParam() {
277284
INSTANCE.outputDirectory = "output-directory";
278285
INSTANCE.help = false;
@@ -324,6 +331,8 @@ public static void clearParam() {
324331
INSTANCE.walletExtensionApi = false;
325332
INSTANCE.connectFactor = 0.3;
326333
INSTANCE.activeConnectFactor = 0.1;
334+
INSTANCE.disconnectNumberFactor = 0.4;
335+
INSTANCE.maxConnectNumberFactor = 0.8;
327336
}
328337

329338
/**
@@ -545,6 +554,11 @@ public static void setParam(final String[] args, final String confFileName) {
545554
INSTANCE.activeConnectFactor = config.hasPath("node.activeConnectFactor") ?
546555
config.getDouble("node.activeConnectFactor") : 0.1;
547556

557+
INSTANCE.disconnectNumberFactor = config.hasPath("node.disconnectNumberFactor") ?
558+
config.getDouble("node.disconnectNumberFactor") : 0.4;
559+
INSTANCE.maxConnectNumberFactor = config.hasPath("node.maxConnectNumberFactor") ?
560+
config.getDouble("node.maxConnectNumberFactor") : 0.8;
561+
548562
initBackupProperty(config);
549563

550564
logConfig();

‎src/main/resources/config.conf‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,10 @@ node {
9090

9191
minParticipationRate = 15
9292

93+
# check the peer data transfer ,disconnect factor
94+
disconnectNumberFactor = 0.4
95+
maxConnectNumberFactor = 0.8
96+
9397
p2p {
9498
version = 11111 # 11111: mainnet; 20180622: testnet
9599
}

0 commit comments

Comments
 (0)