Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improve threadpools at networknode #6458

Merged
merged 5 commits into from
Dec 14, 2022
Merged

Improve threadpools at networknode #6458

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d4b6965
Use networkNode.sendMessage instead of connection.sendMessage as the …
HenrikJannsen Dec 13, 2022
0dc8d0b
Add maxConnection parameter to be used for setting the core pool size…
HenrikJannsen Dec 13, 2022
84584d3
The connection:keep-alive header is restricted in java and would requ…
HenrikJannsen Dec 13, 2022
071a352
Remove rejectedExecutionHandler at newCachedThreadPool.
HenrikJannsen Dec 13, 2022
f90c4b8
Use cachedThreadPool which uses SynchronousQueue instead of ArrayBloc…
HenrikJannsen Dec 13, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 2 additions & 5 deletions common/src/main/java/bisq/common/util/Utilities.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,6 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
Expand Down Expand Up @@ -139,14 +138,12 @@ private static ThreadPoolExecutor getThreadPoolExecutor(String name,

public static ExecutorService newCachedThreadPool(int maximumPoolSize,
long keepAliveTime,
TimeUnit timeUnit,
RejectedExecutionHandler rejectedExecutionHandler) {
TimeUnit timeUnit) {
return new ThreadPoolExecutor(0,
maximumPoolSize,
keepAliveTime,
timeUnit,
new SynchronousQueue<>(),
rejectedExecutionHandler);
new SynchronousQueue<>());
}

@SuppressWarnings("SameParameterValue")
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/java/bisq/core/dao/monitoring/network/StateNetworkService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ public void sendGetStateHashesResponse(Connection connection, int nonce, List<St
Res getStateHashesResponse = getGetStateHashesResponse(nonce, stateHashes);
log.info("Send {} with {} stateHashes to peer {}", getStateHashesResponse.getClass().getSimpleName(),
stateHashes.size(), connection.getPeersNodeAddressOptional());
connection.sendMessage(getStateHashesResponse);
networkNode.sendMessage(connection, getStateHashesResponse);
}

public void requestHashesFromAllConnectedSeedNodes(int fromHeight) {
Expand Down
5 changes: 3 additions & 2 deletions p2p/src/main/java/bisq/network/p2p/NetworkNodeProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ public class NetworkNodeProvider implements Provider<NetworkNode> {
public NetworkNodeProvider(NetworkProtoResolver networkProtoResolver,
BridgeAddressProvider bridgeAddressProvider,
@Nullable NetworkFilter networkFilter,
@Named(Config.MAX_CONNECTIONS) int maxConnections,
@Named(Config.USE_LOCALHOST_FOR_P2P) boolean useLocalhostForP2P,
@Named(Config.NODE_PORT) int port,
@Named(Config.TOR_DIR) File torDir,
Expand All @@ -56,7 +57,7 @@ public NetworkNodeProvider(NetworkProtoResolver networkProtoResolver,
@Named(Config.TOR_STREAM_ISOLATION) boolean streamIsolation,
@Named(Config.TOR_CONTROL_USE_SAFE_COOKIE_AUTH) boolean useSafeCookieAuthentication) {
if (useLocalhostForP2P) {
networkNode = new LocalhostNetworkNode(port, networkProtoResolver, networkFilter);
networkNode = new LocalhostNetworkNode(port, networkProtoResolver, networkFilter, maxConnections);
} else {
TorMode torMode = getTorMode(bridgeAddressProvider,
torDir,
Expand All @@ -66,7 +67,7 @@ public NetworkNodeProvider(NetworkProtoResolver networkProtoResolver,
password,
cookieFile,
useSafeCookieAuthentication);
networkNode = new TorNetworkNode(port, networkProtoResolver, streamIsolation, torMode, networkFilter);
networkNode = new TorNetworkNode(port, networkProtoResolver, streamIsolation, torMode, networkFilter, maxConnections);
}
}

Expand Down
5 changes: 3 additions & 2 deletions p2p/src/main/java/bisq/network/p2p/network/LocalhostNetworkNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,8 +56,9 @@ public static void setSimulateTorDelayHiddenService(int simulateTorDelayHiddenSe

public LocalhostNetworkNode(int port,
NetworkProtoResolver networkProtoResolver,
@Nullable NetworkFilter networkFilter) {
super(port, networkProtoResolver, networkFilter);
@Nullable NetworkFilter networkFilter,
int maxConnections) {
super(port, networkProtoResolver, networkFilter, maxConnections);
}

@Override
Expand Down
10 changes: 7 additions & 3 deletions p2p/src/main/java/bisq/network/p2p/network/NetworkNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,7 @@ public abstract class NetworkNode implements MessageListener {
private final NetworkProtoResolver networkProtoResolver;
@Nullable
private final NetworkFilter networkFilter;
private final int maxConnections;

private final CopyOnWriteArraySet<InboundConnection> inBoundConnections = new CopyOnWriteArraySet<>();
private final CopyOnWriteArraySet<MessageListener> messageListeners = new CopyOnWriteArraySet<>();
Expand All @@ -92,10 +93,12 @@ public abstract class NetworkNode implements MessageListener {

NetworkNode(int servicePort,
NetworkProtoResolver networkProtoResolver,
@Nullable NetworkFilter networkFilter) {
@Nullable NetworkFilter networkFilter,
int maxConnections) {
this.servicePort = servicePort;
this.networkProtoResolver = networkProtoResolver;
this.networkFilter = networkFilter;
this.maxConnections = maxConnections;
}

///////////////////////////////////////////////////////////////////////////////////////////
Expand Down Expand Up @@ -436,8 +439,9 @@ public void removeMessageListener(MessageListener messageListener) {
///////////////////////////////////////////////////////////////////////////////////////////

void createExecutorService() {
if (executorService == null)
executorService = Utilities.getListeningExecutorService("NetworkNode-" + servicePort, 15, 30, 60);
if (executorService == null) {
executorService = MoreExecutors.listeningDecorator(Utilities.newCachedThreadPool(maxConnections * 4, 3, TimeUnit.MINUTES));
}
}

void startServer(ServerSocket serverSocket) {
Expand Down
5 changes: 3 additions & 2 deletions p2p/src/main/java/bisq/network/p2p/network/TorNetworkNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,8 +78,9 @@ public TorNetworkNode(int servicePort,
NetworkProtoResolver networkProtoResolver,
boolean useStreamIsolation,
TorMode torMode,
@Nullable NetworkFilter networkFilter) {
super(servicePort, networkProtoResolver, networkFilter);
@Nullable NetworkFilter networkFilter,
int maxConnections) {
super(servicePort, networkProtoResolver, networkFilter, maxConnections);
this.torMode = torMode;
this.streamIsolation = useStreamIsolation;
createExecutorService();
Expand Down
4 changes: 2 additions & 2 deletions p2p/src/test/java/bisq/network/p2p/network/LocalhostNetworkNodeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ public class LocalhostNetworkNodeTest {
@Test
public void testMessage() throws InterruptedException, IOException {
CountDownLatch msgLatch = new CountDownLatch(2);
LocalhostNetworkNode node1 = new LocalhostNetworkNode(9001, TestUtils.getNetworkProtoResolver(), null);
LocalhostNetworkNode node1 = new LocalhostNetworkNode(9001, TestUtils.getNetworkProtoResolver(), null, 12);
node1.addMessageListener((message, connection) -> {
log.debug("onMessage node1 " + message);
msgLatch.countDown();
Expand Down Expand Up @@ -69,7 +69,7 @@ public void onRequestCustomBridges() {
}
});

LocalhostNetworkNode node2 = new LocalhostNetworkNode(9002, TestUtils.getNetworkProtoResolver(), null);
LocalhostNetworkNode node2 = new LocalhostNetworkNode(9002, TestUtils.getNetworkProtoResolver(), null, 12);
node2.addMessageListener((message, connection) -> {
log.debug("onMessage node2 " + message);
msgLatch.countDown();
Expand Down
8 changes: 4 additions & 4 deletions p2p/src/test/java/bisq/network/p2p/network/TorNetworkNodeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ public void testTorNodeBeforeSecondReady() throws InterruptedException, IOExcept
latch = new CountDownLatch(1);
int port = 9001;
TorNetworkNode node1 = new TorNetworkNode(port, TestUtils.getNetworkProtoResolver(), false,
new NewTor(new File("torNode_" + port), null, "", this::getBridgeAddresses), null);
new NewTor(new File("torNode_" + port), null, "", this::getBridgeAddresses), null, 12);
node1.start(new SetupListener() {
@Override
public void onTorNodeReady() {
Expand All @@ -83,7 +83,7 @@ public void onRequestCustomBridges() {
latch = new CountDownLatch(1);
int port2 = 9002;
TorNetworkNode node2 = new TorNetworkNode(port2, TestUtils.getNetworkProtoResolver(), false,
new NewTor(new File("torNode_" + port), null, "", this::getBridgeAddresses), null);
new NewTor(new File("torNode_" + port), null, "", this::getBridgeAddresses), null, 12);
node2.start(new SetupListener() {
@Override
public void onTorNodeReady() {
Expand Down Expand Up @@ -141,7 +141,7 @@ public void testTorNodeAfterBothReady() throws InterruptedException, IOException
latch = new CountDownLatch(2);
int port = 9001;
TorNetworkNode node1 = new TorNetworkNode(port, TestUtils.getNetworkProtoResolver(), false,
new NewTor(new File("torNode_" + port), null, "", this::getBridgeAddresses), null);
new NewTor(new File("torNode_" + port), null, "", this::getBridgeAddresses), null, 12);
node1.start(new SetupListener() {
@Override
public void onTorNodeReady() {
Expand All @@ -167,7 +167,7 @@ public void onRequestCustomBridges() {

int port2 = 9002;
TorNetworkNode node2 = new TorNetworkNode(port2, TestUtils.getNetworkProtoResolver(), false,
new NewTor(new File("torNode_" + port), null, "", this::getBridgeAddresses), null);
new NewTor(new File("torNode_" + port), null, "", this::getBridgeAddresses), null, 12);
node2.start(new SetupListener() {
@Override
public void onTorNodeReady() {
Expand Down
14 changes: 7 additions & 7 deletions seednode/src/main/java/bisq/seednode/reporting/SeedNodeReportingService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;

Expand Down Expand Up @@ -115,10 +116,9 @@ public SeedNodeReportingService(P2PService p2PService,
this.maxConnections = maxConnections;
this.seedNodeReportingServerUrl = seedNodeReportingServerUrl;

executor = Utilities.newCachedThreadPool(5,
8,
TimeUnit.MINUTES,
(runnable, executor) -> log.error("Execution was rejected. We skip the {} task.", runnable.toString()));
// The pool size must be larger as the expected parallel sends because HttpClient use it
// internally for asynchronous and dependent tasks.
executor = Utilities.newCachedThreadPool(20, 8, TimeUnit.MINUTES);
httpClient = HttpClient.newBuilder().executor(executor).build();

heartBeatTimer = UserThread.runPeriodically(this::sendHeartBeat, HEART_BEAT_DELAY_SEC);
Expand Down Expand Up @@ -249,7 +249,6 @@ private void sendReportingItems(ReportingItems reportingItems) {
.uri(URI.create(seedNodeReportingServerUrl))
.POST(HttpRequest.BodyPublishers.ofByteArray(protoMessageAsBytes))
.header("User-Agent", getMyAddress())
.header("Connection", "keep-alive")
.build();
httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()).whenComplete((response, throwable) -> {
if (throwable != null) {
Expand All @@ -258,9 +257,10 @@ private void sendReportingItems(ReportingItems reportingItems) {
log.error("Response error message: {}", response);
}
});
} catch (RejectedExecutionException t) {
log.warn("Did not send reportingItems {} because of RejectedExecutionException {}", reportingItems, t.toString());
} catch (Throwable t) {
// RejectedExecutionException is thrown if we exceed our pool size.
log.error("Did not send reportingItems {} because of exception {}", reportingItems, t.toString());
log.warn("Did not send reportingItems {} because of exception {}", reportingItems, t.toString());
}
}

Expand Down