Skip to content

Commit 99cce7e

Browse files
joaomlnetojoaomlneto
authored
Merge pull request #160 from joaomlneto/hBFT-20250225
Cleanup hBFT
2 parents 9a28022 + 5fa187c commit 99cce7e

14 files changed

+374
-570
lines changed

‎simulator/src/main/java/byzzbench/simulator/BaseScenario.java

-13
Original file line numberDiff line numberDiff line change
@@ -127,19 +127,6 @@ protected void setNumClients(int numClients) {
127127
}
128128
}
129129

130-
/**
131-
* Sets the number of hBFT clients in the scenario.
132-
*
133-
* @param numClients The number of clients to set.
134-
*/
135-
protected void setNumHbftClients(int numClients) {
136-
for (int i = 0; i < numClients; i++) {
137-
String clientId = String.format("C%d", i);
138-
Client client = HbftClient.builder().id(clientId).scenario(this).build();
139-
this.addClient(client);
140-
}
141-
}
142-
143130
/**
144131
* Adds a client to the scenario.
145132
*

‎simulator/src/main/java/byzzbench/simulator/Client.java

+5-5
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
package byzzbench.simulator;
22

3+
import byzzbench.simulator.protocols.hbft.message.ClientRequestMessage;
34
import byzzbench.simulator.transport.MessagePayload;
45
import byzzbench.simulator.utils.NonNull;
56
import com.fasterxml.jackson.annotation.JsonIgnore;
6-
import lombok.Builder;
77
import lombok.Getter;
88
import lombok.RequiredArgsConstructor;
99
import lombok.experimental.SuperBuilder;
@@ -64,8 +64,8 @@ public void initialize() {
6464
*/
6565
public void sendRequest() {
6666
String recipientId = this.getScenario().getReplicas().keySet().iterator().next();
67-
String requestId = String.format("%s/%d", this.id, this.requestSequenceNumber.getAndIncrement());
68-
this.getScenario().getTransport().sendClientRequest(this.id, requestId, recipientId);
67+
MessagePayload payload = new ClientRequestMessage(this.getCurrentTime().toEpochMilli(), recipientId);
68+
this.getScenario().getTransport().sendMessage(this, payload, recipientId);
6969
}
7070

7171
/**
@@ -95,7 +95,7 @@ public Instant getCurrentTime() {
9595
* @return the timer object
9696
*/
9797
public long setTimeout(String name, Runnable r, Duration timeout) {
98-
return this.scenario.getTransport().setTimeout(this, r, timeout);
98+
return this.scenario.getTransport().setTimeout(this, r, timeout, "REQUEST");
9999
}
100100

101101
/**
@@ -113,4 +113,4 @@ public void clearTimeout(long eventId) {
113113
public void clearAllTimeouts() {
114114
this.scenario.getTransport().clearReplicaTimeouts(this);
115115
}
116-
}
116+
}

‎simulator/src/main/java/byzzbench/simulator/HbftClient.java

+29-62
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,17 @@
11
package byzzbench.simulator;
22

3-
import byzzbench.simulator.protocols.hbft.message.PanicMessage;
4-
import byzzbench.simulator.protocols.hbft.message.RequestMessage;
3+
import byzzbench.simulator.protocols.hbft.message.*;
4+
import byzzbench.simulator.protocols.hbft.pojo.ClientReplyKey;
55
import byzzbench.simulator.transport.MessagePayload;
6-
76
import com.fasterxml.jackson.annotation.JsonIgnore;
8-
97
import lombok.Getter;
108
import lombok.experimental.SuperBuilder;
119

1210
import java.io.Serializable;
1311
import java.security.MessageDigest;
1412
import java.security.NoSuchAlgorithmException;
1513
import java.time.Duration;
16-
import java.util.ArrayList;
17-
import java.util.Collection;
18-
import java.util.HashSet;
19-
import java.util.Set;
20-
import java.util.SortedMap;
21-
import java.util.TreeMap;
22-
23-
import byzzbench.simulator.protocols.hbft.message.ReplyMessage;
24-
import byzzbench.simulator.protocols.hbft.pojo.ClientReplyKey;
14+
import java.util.*;
2515

2616

2717
/**
@@ -82,11 +72,11 @@ public class HbftClient extends Client {
8272
@Override
8373
public void sendRequest() {
8474
String requestId = String.format("%s/%d", super.id, super.requestSequenceNumber.incrementAndGet());
85-
long timestamp = System.currentTimeMillis();
75+
long timestamp = this.getCurrentTime().toEpochMilli();
8676
RequestMessage request = new RequestMessage(requestId, timestamp, super.id);
8777
this.sentRequests.put(super.requestSequenceNumber.get(), request);
8878
this.sentRequestsByTimestamp.put(timestamp, requestId);
89-
super.getScenario().getTransport().multicastClientRequest(super.id, timestamp, requestId, super.scenario.getTransport().getNodeIds());
79+
this.broadcastRequest(timestamp, requestId);
9080

9181
// Set timeout
9282
Long timeoutId = this.setTimeout("REQUEST", this::retransmitOrPanic, this.timeout);
@@ -100,30 +90,37 @@ public void retransmitOrPanic() {
10090
// Based on hBFT 4.1 it uses the identical request
10191
// TODO: It probably should not be the same timestamp
10292
long timestamp = this.sentRequests.get(super.requestSequenceNumber.get()).getTimestamp();
103-
super.scenario.getTransport().multicastClientRequest(super.id, timestamp, requestId, super.scenario.getTransport().getNodeIds());
93+
this.broadcastRequest(timestamp, requestId);
10494
} else if (this.shouldPanic(tolerance)) {
10595
RequestMessage message = this.sentRequests.get(super.requestSequenceNumber.get());
106-
PanicMessage panic = new PanicMessage(this.digest(message), System.currentTimeMillis(), super.id);
96+
PanicMessage panic = new PanicMessage(this.digest(message), this.getCurrentTime().toEpochMilli(), super.id);
10797
super.scenario.getTransport().multicast(this, super.scenario.getTransport().getNodeIds(), panic);
10898
}
10999
this.clearTimeout(timeouts.get(super.requestSequenceNumber.get()));
110100
Long timeoutId = this.setTimeout("REQUEST", this::retransmitOrPanic, this.timeout);
111101
timeouts.put(super.requestSequenceNumber.get(), timeoutId);
112102
}
113103

104+
private void broadcastRequest(long timestamp, String requestId) {
105+
MessagePayload payload = new ClientRequestMessage(timestamp, requestId);
106+
SortedSet<String> replicaIds = super.scenario.getTransport().getNodeIds();
107+
getScenario().getTransport().multicast(this, replicaIds, payload);
108+
}
109+
114110
/**
115111
* Handles a reply received by the client.
112+
*
116113
* @param senderId The ID of the sender of the reply.
117-
* @param reply The reply received by the client.
118-
* @param tolerance the tolerance of the protocol (used for hbft)
114+
* @param payload The payload received by the client.
119115
*/
120-
public void handleReply(String senderId, MessagePayload reply, long tolerance, long seqNumber) {
121-
if (!(reply instanceof ReplyMessage)) {
116+
public void handleMessage(String senderId, MessagePayload payload) {
117+
if (!(payload instanceof ClientReplyMessage clientReplyMessage)) {
122118
return;
123119
}
124-
ClientReplyKey key = new ClientReplyKey(((ReplyMessage) reply).getResult().toString(), seqNumber);
120+
ReplyMessage reply = clientReplyMessage.getReply();
121+
ClientReplyKey key = new ClientReplyKey(reply.getResult().toString(), reply.getSequenceNumber());
125122
// Default is for testing only
126-
String currRequest = this.sentRequestsByTimestamp.getOrDefault(((ReplyMessage) reply).getTimestamp(), "C/0");
123+
String currRequest = this.sentRequestsByTimestamp.getOrDefault(reply.getTimestamp(), "C/0");
127124
this.hbftreplies.putIfAbsent(currRequest, new TreeMap<>());
128125
this.hbftreplies.get(currRequest).putIfAbsent(key, new ArrayList<>());
129126
this.hbftreplies.get(currRequest).get(key).add(reply);
@@ -132,30 +129,15 @@ public void handleReply(String senderId, MessagePayload reply, long tolerance, l
132129
* If the client received 2f + 1 correct replies,
133130
* and the request has not been completed yet.
134131
*/
135-
if (this.completedReplies(tolerance)
136-
&& !this.completedRequests.contains(key)
137-
&& super.requestSequenceNumber.get() <= this.maxRequests) {
138-
this.completedRequests.add(key);
139-
this.clearTimeout(this.timeouts.get(super.requestSequenceNumber.get()));
140-
this.sendRequest();
132+
if (this.completedReplies(clientReplyMessage.getTolerance())
133+
&& !this.completedRequests.contains(key)
134+
&& super.requestSequenceNumber.get() <= this.maxRequests) {
135+
this.completedRequests.add(key);
136+
this.clearTimeout(this.timeouts.get(super.requestSequenceNumber.get()));
137+
this.sendRequest();
141138
}
142139
}
143140

144-
/**
145-
* Handles a reply received by the client.
146-
*
147-
* @param senderId The ID of the sender of the reply.
148-
* @param reply The reply received by the client.
149-
*/
150-
@Override
151-
public void handleMessage(String senderId, MessagePayload reply) {
152-
// this.replies.putIfAbsent(super.requestSequenceNumber.get(), new ArrayList<>());
153-
// this.replies.get(super.requestSequenceNumber.get()).add(reply);
154-
// if (super.requestSequenceNumber.get() < this.maxRequests) {
155-
// this.sendRequest();
156-
// }
157-
}
158-
159141
/**
160142
* Set a timeout for this replica.
161143
*
@@ -166,22 +148,7 @@ public void handleMessage(String senderId, MessagePayload reply) {
166148
*/
167149
public long setTimeout(String name, Runnable r, long timeout) {
168150
Duration duration = Duration.ofSeconds(timeout);
169-
return super.scenario.getTransport().setTimeout(this, r, duration);
170-
}
171-
172-
/**
173-
* Set a timeout for this client.
174-
*
175-
* @param r the runnable to execute when the timeout occurs
176-
* @param timeout the timeout in milliseconds
177-
* @return the timeout ID
178-
*/
179-
public long setTimeout(Runnable r, long timeout) {
180-
Runnable wrapper = () -> {
181-
r.run();
182-
};
183-
Duration duration = Duration.ofSeconds(timeout);
184-
return super.scenario.getTransport().setClientTimeout(super.id, wrapper, duration);
151+
return super.scenario.getTransport().setTimeout(this, r, duration, name);
185152
}
186153

187154
/**
@@ -206,8 +173,8 @@ public boolean shouldRetransmit(long tolerance) {
206173
public boolean shouldPanic(long tolerance) {
207174
String currRequest = String.format("%s/%d", super.id, super.requestSequenceNumber.get());
208175
for (ClientReplyKey key : hbftreplies.get(currRequest).keySet()) {
209-
return this.hbftreplies.get(currRequest).get(key).size() >= tolerance + 1
210-
&& this.hbftreplies.get(currRequest).get(key).size() < tolerance * 2 + 1;
176+
return this.hbftreplies.get(currRequest).get(key).size() >= tolerance + 1
177+
&& this.hbftreplies.get(currRequest).get(key).size() < tolerance * 2 + 1;
211178
}
212179
return false;
213180
}

‎simulator/src/main/java/byzzbench/simulator/Replica.java

+3-40
Original file line numberDiff line numberDiff line change
@@ -164,8 +164,8 @@ public void initialize() {
164164
/**
165165
* Handle a request received from a client.
166166
*
167-
* @param clientId the ID of the client
168-
* @param request the request payload
167+
* @param clientId the ID of the client
168+
* @param request the request payload
169169
* @param timestamp the time the request was created/sent
170170
* @throws Exception if an error occurs while handling the request
171171
*/
@@ -181,26 +181,6 @@ public void sendReplyToClient(String clientId, Serializable reply) {
181181
this.transport.sendClientResponse(this, new DefaultClientReplyPayload(reply), clientId);
182182
}
183183

184-
/**
185-
* Send a reply to a client.
186-
* @param clientId the ID of the client
187-
* @param reply the reply payload
188-
* @param tolerance the tolerance of the protocol (used for hbft)
189-
*/
190-
public void sendReplyToClient(String clientId, MessagePayload reply, long tolerance, long seqNumber) {
191-
this.scenario.getTransport().sendClientResponse(this.id, reply, clientId, tolerance, seqNumber);
192-
}
193-
194-
/**
195-
* Handle a message received by this replica.
196-
*
197-
* @param sender the ID of the sender
198-
* @param message the message payload
199-
* @throws Exception if an error occurs while handling the message
200-
*/
201-
public abstract void handleMessage(String sender, MessagePayload message)
202-
throws Exception;
203-
204184
/**
205185
* Commit an operation to the commit log and notify observers.
206186
*
@@ -237,7 +217,7 @@ public long setTimeout(String name, Runnable r, Duration timeout) {
237217
this.notifyObserversTimeout();
238218
r.run();
239219
};
240-
return this.transport.setTimeout(this, wrapper, timeout);
220+
return this.transport.setTimeout(this, wrapper, timeout, name);
241221
}
242222

243223
/**
@@ -249,23 +229,6 @@ public void clearTimeout(long eventId) {
249229
this.transport.clearTimeout(this, eventId);
250230
}
251231

252-
/**
253-
* Set a timeout for this replica.
254-
*
255-
* @param r the runnable to execute when the timeout occurs
256-
* @param timeout the timeout in milliseconds
257-
* @param description the type of timeout
258-
* @return the timeout ID
259-
*/
260-
public long setTimeout(Runnable r, long timeout, String description) {
261-
Runnable wrapper = () -> {
262-
this.notifyObserversTimeout();
263-
r.run();
264-
};
265-
Duration duration = Duration.ofSeconds(timeout);
266-
return this.scenario.getTransport().setTimeout(this, wrapper, duration, description);
267-
}
268-
269232
/**
270233
* Clear timeout based on description.
271234
*/

‎simulator/src/main/java/byzzbench/simulator/protocols/XRPL/XRPLScenario.java

+24-5
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import byzzbench.simulator.BaseScenario;
44
import byzzbench.simulator.Client;
55
import byzzbench.simulator.scheduler.Scheduler;
6+
import byzzbench.simulator.transport.DefaultClientRequestPayload;
67
import com.fasterxml.jackson.databind.JsonNode;
78

89
import java.util.ArrayList;
@@ -105,8 +106,8 @@ private void setupForScenario3() {
105106
@SuppressWarnings("unused")
106107
private void runScenario1() {
107108
try {
108-
this.transport.sendClientRequest("C0", "0000", "A");
109-
this.transport.sendClientRequest("C0", "0001", "B");
109+
this.sendClientRequest("C0", "0000", "A");
110+
this.sendClientRequest("C0", "0001", "B");
110111

111112
this.initializeHeartbeats();
112113
} catch (Exception e) {
@@ -121,7 +122,7 @@ private void runScenario1() {
121122
@SuppressWarnings("unused")
122123
private void runScenario2() {
123124
try {
124-
this.transport.sendClientRequest("c1", "0000", "A");
125+
this.sendClientRequest("c1", "0000", "A");
125126

126127
this.initializeHeartbeats();
127128
} catch (Exception e) {
@@ -141,10 +142,15 @@ private void runScenario3() {
141142
this.addClient(new Client(this, "C0") {
142143
@Override
143144
public void initialize() {
144-
this.getScenario().getTransport().sendClientRequest(this.getId(), "tx", "D");
145+
this.getScenario().getTransport().sendMessage(
146+
this,
147+
new DefaultClientRequestPayload("tx"),
148+
"D"
149+
);
150+
//this.getScenario().getTransport().sendClientRequest(this.getId(), "tx", "D");
145151
}
146152
});
147-
this.transport.sendClientRequest("C0", "tx", "D");
153+
this.sendClientRequest("C0", "tx", "D");
148154
this.initializeHeartbeats();
149155
} catch (Exception e) {
150156
e.printStackTrace();
@@ -167,4 +173,17 @@ public int maxFaultyReplicas(int n) {
167173
}
168174
return maxFaultyReplicas;
169175
}
176+
177+
/**
178+
* Sends a client request to a replica in the system.
179+
*
180+
* @param fromId The ID of the client sending the request
181+
* @param operation The operation to be performed
182+
* @param toId The ID of the replica receiving the request
183+
*/
184+
private void sendClientRequest(String fromId, String operation, String toId) {
185+
Client from = this.getClients().get(fromId);
186+
DefaultClientRequestPayload payload = new DefaultClientRequestPayload(operation);
187+
this.transport.sendMessage(from, payload, toId);
188+
}
170189
}

0 commit comments

Comments
 (0)