Skip to content

Commit 4214776

Browse files
metanetmdogan
metanet
authored and
mdogan
committed
Store retried wait keys in a single container
When a wait key is expired, all its retried duplicates must expire as well.
1 parent b79c559 commit 4214776

53 files changed

Lines changed: 1141 additions & 540 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎hazelcast-client/src/main/java/com/hazelcast/client/ClientOutOfMemoryHandler.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,7 @@ private static void closeSockets(HazelcastClientInstanceImpl client) {
6565

6666
private static void tryShutdown(HazelcastClientInstanceImpl client) {
6767
try {
68-
client.doShutdown();
68+
client.doShutdown(false);
6969
} catch (Throwable ignored) {
7070
ignore(ignored);
7171
}

‎hazelcast-client/src/main/java/com/hazelcast/client/cp/internal/datastructures/countdownlatch/RaftCountDownLatchProxy.java‎

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@
3131
import com.hazelcast.nio.Bits;
3232
import com.hazelcast.spi.InternalCompletableFuture;
3333
import com.hazelcast.util.EmptyStatement;
34-
import com.hazelcast.util.UuidUtil;
3534

3635
import java.util.UUID;
3736
import java.util.concurrent.TimeUnit;
@@ -45,6 +44,7 @@
4544
import static com.hazelcast.cp.internal.datastructures.countdownlatch.client.CountDownLatchMessageTaskFactoryProvider.GET_ROUND_TYPE;
4645
import static com.hazelcast.cp.internal.datastructures.countdownlatch.client.CountDownLatchMessageTaskFactoryProvider.TRY_SET_COUNT_TYPE;
4746
import static com.hazelcast.util.Preconditions.checkNotNull;
47+
import static com.hazelcast.util.UuidUtil.newUnsecureUUID;
4848

4949
/**
5050
* Client-side Raft-based proxy implementation of {@link ICountDownLatch}
@@ -67,10 +67,14 @@ class RaftCountDownLatchProxy extends ClientProxy implements ICountDownLatch {
6767
public boolean await(long timeout, TimeUnit unit) {
6868
checkNotNull(unit);
6969

70+
UUID invocationUid = newUnsecureUUID();
7071
long timeoutMillis = Math.max(0, unit.toMillis(timeout));
7172

72-
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(objectName) + Bits.LONG_SIZE_IN_BYTES;
73+
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(objectName)
74+
+ Bits.LONG_SIZE_IN_BYTES * 3;
7375
ClientMessage msg = prepareClientMessage(groupId, objectName, dataSize, AWAIT_TYPE);
76+
msg.set(invocationUid.getLeastSignificantBits());
77+
msg.set(invocationUid.getMostSignificantBits());
7478
msg.set(timeoutMillis);
7579
msg.updateFrameLength();
7680

@@ -80,13 +84,13 @@ public boolean await(long timeout, TimeUnit unit) {
8084
@Override
8185
public void countDown() {
8286
int round = getRound();
83-
UUID invocationUid = UuidUtil.newUnsecureUUID();
87+
UUID invocationUid = newUnsecureUUID();
8488
for (;;) {
8589
try {
8690
countDown(round, invocationUid);
8791
return;
88-
} catch (OperationTimeoutException ignored) {
89-
EmptyStatement.ignore(ignored);
92+
} catch (OperationTimeoutException e) {
93+
EmptyStatement.ignore(e);
9094
// I can retry safely because my retry would be idempotent...
9195
}
9296
}
@@ -101,12 +105,12 @@ private int getRound() {
101105
}
102106

103107
private void countDown(int round, UUID invocationUid) {
104-
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(objectName) + Bits.INT_SIZE_IN_BYTES
105-
+ Bits.LONG_SIZE_IN_BYTES * 2;
108+
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(objectName) + Bits.LONG_SIZE_IN_BYTES * 2
109+
+ Bits.INT_SIZE_IN_BYTES;
106110
ClientMessage msg = prepareClientMessage(groupId, objectName, dataSize, COUNT_DOWN_TYPE);
107-
msg.set(round);
108111
msg.set(invocationUid.getLeastSignificantBits());
109112
msg.set(invocationUid.getMostSignificantBits());
113+
msg.set(round);
110114
msg.updateFrameLength();
111115

112116
invoke(msg, INT_RESPONSE_DECODER).join();

‎hazelcast-client/src/main/java/com/hazelcast/client/impl/clientside/HazelcastClientInstanceImpl.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -960,8 +960,10 @@ public void shutdown() {
960960
getLifecycleService().shutdown();
961961
}
962962

963-
public void doShutdown() {
964-
proxySessionManager.shutdown();
963+
public void doShutdown(boolean isGraceful) {
964+
if (isGraceful) {
965+
proxySessionManager.shutdown();
966+
}
965967
proxyManager.destroy();
966968
connectionManager.shutdown();
967969
clusterService.shutdown();

‎hazelcast-client/src/main/java/com/hazelcast/client/impl/clientside/LifecycleServiceImpl.java‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -143,23 +143,27 @@ public boolean isRunning() {
143143

144144
@Override
145145
public void shutdown() {
146+
doShutdown(true);
147+
}
148+
149+
@Override
150+
public void terminate() {
151+
doShutdown(false);
152+
}
153+
154+
private void doShutdown(boolean isGraceful) {
146155
if (!active.compareAndSet(true, false)) {
147156
return;
148157
}
149158

150159
fireLifecycleEvent(SHUTTING_DOWN);
151160
HazelcastClient.shutdown(client.getName());
152-
client.doShutdown();
161+
client.doShutdown(isGraceful);
153162
fireLifecycleEvent(SHUTDOWN);
154163

155164
shutdownExecutor();
156165
}
157166

158-
@Override
159-
public void terminate() {
160-
shutdown();
161-
}
162-
163167
private void shutdownExecutor() {
164168
executor.shutdown();
165169
try {

‎hazelcast/src/main/java/com/hazelcast/cp/internal/MetadataRaftGroupManager.java‎

Lines changed: 48 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -426,6 +426,10 @@ public void triggerRemoveMember(CPMember leavingMember) {
426426
return;
427427
}
428428

429+
initializeMembershipChangeContextForLeavingMember(leavingMember);
430+
}
431+
432+
private void initializeMembershipChangeContextForLeavingMember(CPMember leavingMember) {
429433
List<CPGroupId> leavingGroupIds = new ArrayList<CPGroupId>();
430434
List<CPGroupMembershipChangeContext> leavingGroups = new ArrayList<CPGroupMembershipChangeContext>();
431435
for (RaftGroup group : groups.values()) {
@@ -439,16 +443,10 @@ public void triggerRemoveMember(CPMember leavingMember) {
439443
leavingGroupIds.add(groupId);
440444
leavingGroups.add(new CPGroupMembershipChangeContext(groupId, group.getMembersCommitIndex(),
441445
group.memberImpls(), substitute, leavingMember));
442-
if (logger.isFineEnabled()) {
443-
logger.fine("Substituted " + leavingMember + " with " + substitute + " in " + group);
444-
}
445446
} else {
446447
leavingGroupIds.add(groupId);
447448
leavingGroups.add(new CPGroupMembershipChangeContext(groupId, group.getMembersCommitIndex(),
448449
group.memberImpls(), null, leavingMember));
449-
if (logger.isFineEnabled()) {
450-
logger.fine("Could not find a substitute for " + leavingMember + " in " + group);
451-
}
452450
}
453451
}
454452

@@ -760,12 +758,7 @@ private class DiscoverInitialCPMembersTask implements Runnable {
760758

761759
@Override
762760
public void run() {
763-
if (isDiscoveryCompleted()) {
764-
return;
765-
}
766-
767-
if (!nodeEngine.getClusterService().isJoined()) {
768-
scheduleDiscoveryInitialCPMembersTask();
761+
if (shouldSkipOrReschedule()) {
769762
return;
770763
}
771764

@@ -777,25 +770,16 @@ public void run() {
777770
return;
778771
}
779772
}
773+
780774
latestMembers = members;
781775

782-
if (members.size() < config.getCPMemberCount()) {
783-
if (logger.isFineEnabled()) {
784-
logger.warning("Waiting for " + config.getCPMemberCount() + " CP members to join the cluster. "
785-
+ "Current CP member count: " + members.size());
786-
}
787-
scheduleDiscoveryInitialCPMembersTask();
776+
if (rescheduleIfCPMemberCountNotSatisfied(members)) {
788777
return;
789778
}
790779

791780
List<CPMember> cpMembers = getInitialCPMembers(members);
792781

793-
if (!cpMembers.contains(getLocalMember())) {
794-
if (logger.isFineEnabled()) {
795-
logger.fine("I am not an initial CP member! I'll serve as an AP member.");
796-
}
797-
localMember.set(null);
798-
disableDiscovery();
782+
if (completeDiscoveryIfNotCPMember(cpMembers)) {
799783
return;
800784
}
801785

@@ -813,6 +797,46 @@ public void run() {
813797
scheduleRaftGroupMembershipManagementTasks();
814798
}
815799

800+
private boolean shouldSkipOrReschedule() {
801+
if (isDiscoveryCompleted()) {
802+
return true;
803+
}
804+
805+
if (!nodeEngine.getClusterService().isJoined()) {
806+
scheduleDiscoveryInitialCPMembersTask();
807+
return true;
808+
}
809+
810+
return false;
811+
}
812+
813+
private boolean rescheduleIfCPMemberCountNotSatisfied(Collection<Member> members) {
814+
if (members.size() < config.getCPMemberCount()) {
815+
if (logger.isFineEnabled()) {
816+
logger.fine("Waiting for " + config.getCPMemberCount() + " CP members to join the cluster. "
817+
+ "Current CP member count: " + members.size());
818+
}
819+
820+
scheduleDiscoveryInitialCPMembersTask();
821+
return true;
822+
}
823+
return false;
824+
}
825+
826+
private boolean completeDiscoveryIfNotCPMember(List<CPMember> cpMembers) {
827+
if (!cpMembers.contains(getLocalMember())) {
828+
if (logger.isFineEnabled()) {
829+
logger.fine("I am not an initial CP member! I'll serve as an AP member.");
830+
}
831+
832+
localMember.set(null);
833+
disableDiscovery();
834+
return true;
835+
}
836+
837+
return false;
838+
}
839+
816840
@SuppressWarnings("unchecked")
817841
private boolean commitInitialMetadataRaftGroup(List<CPMember> initialCPMembers) {
818842
int metadataGroupSize = config.getGroupSize();

‎hazelcast/src/main/java/com/hazelcast/cp/internal/datastructures/RaftDataServiceDataSerializerHook.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
package com.hazelcast.cp.internal.datastructures;
1818

19+
import com.hazelcast.cp.internal.datastructures.spi.blocking.WaitKeyContainer;
1920
import com.hazelcast.internal.serialization.DataSerializerHook;
2021
import com.hazelcast.internal.serialization.impl.FactoryIdHelper;
2122
import com.hazelcast.nio.serialization.DataSerializableFactory;
@@ -33,8 +34,10 @@ public class RaftDataServiceDataSerializerHook implements DataSerializerHook {
3334

3435
public static final int F_ID = FactoryIdHelper.getFactoryId(RAFT_DS_FACTORY, FACTORY_ID);
3536

36-
public static final int EXPIRE_WAIT_KEYS_OP = 1;
37-
public static final int DESTROY_RAFT_OBJECT_OP = 2;
37+
38+
public static final int WAIT_KEY_CONTAINER = 1;
39+
public static final int EXPIRE_WAIT_KEYS_OP = 2;
40+
public static final int DESTROY_RAFT_OBJECT_OP = 3;
3841

3942
@Override
4043
public int getFactoryId() {
@@ -47,6 +50,8 @@ public DataSerializableFactory createFactory() {
4750
@Override
4851
public IdentifiedDataSerializable create(int typeId) {
4952
switch (typeId) {
53+
case WAIT_KEY_CONTAINER:
54+
return new WaitKeyContainer();
5055
case EXPIRE_WAIT_KEYS_OP:
5156
return new ExpireWaitKeysOp();
5257
case DESTROY_RAFT_OBJECT_OP:

‎hazelcast/src/main/java/com/hazelcast/cp/internal/datastructures/countdownlatch/AwaitInvocationKey.java‎

Lines changed: 19 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -17,37 +17,35 @@
1717
package com.hazelcast.cp.internal.datastructures.countdownlatch;
1818

1919
import com.hazelcast.core.ICountDownLatch;
20+
import com.hazelcast.cp.internal.datastructures.spi.blocking.WaitKey;
2021
import com.hazelcast.nio.ObjectDataInput;
2122
import com.hazelcast.nio.ObjectDataOutput;
2223
import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
23-
import com.hazelcast.cp.internal.datastructures.spi.blocking.WaitKey;
2424

2525
import java.io.IOException;
26+
import java.util.UUID;
2627
import java.util.concurrent.TimeUnit;
2728

2829
import static com.hazelcast.cp.internal.session.AbstractProxySessionManager.NO_SESSION_ID;
30+
import static com.hazelcast.cp.internal.util.UUIDSerializationUtil.readUUID;
31+
import static com.hazelcast.cp.internal.util.UUIDSerializationUtil.writeUUID;
2932
import static com.hazelcast.util.Preconditions.checkNotNull;
3033

3134
/**
3235
* Represents a {@link ICountDownLatch#await(long, TimeUnit)}} invocation
3336
*/
3437
public class AwaitInvocationKey implements WaitKey, IdentifiedDataSerializable {
3538

36-
private String name;
3739
private long commitIndex;
40+
private UUID invocationUid;
3841

3942
AwaitInvocationKey() {
4043
}
4144

42-
AwaitInvocationKey(String name, long commitIndex) {
43-
checkNotNull(name);
44-
this.name = name;
45+
AwaitInvocationKey(long commitIndex, UUID invocationUid) {
46+
checkNotNull(invocationUid);
4547
this.commitIndex = commitIndex;
46-
}
47-
48-
@Override
49-
public String name() {
50-
return name;
48+
this.invocationUid = invocationUid;
5149
}
5250

5351
@Override
@@ -60,6 +58,11 @@ public long commitIndex() {
6058
return commitIndex;
6159
}
6260

61+
@Override
62+
public UUID invocationUid() {
63+
return invocationUid;
64+
}
65+
6366
@Override
6467
public int getFactoryId() {
6568
return RaftCountDownLatchDataSerializerHook.F_ID;
@@ -72,14 +75,14 @@ public int getId() {
7275

7376
@Override
7477
public void writeData(ObjectDataOutput out) throws IOException {
75-
out.writeUTF(name);
7678
out.writeLong(commitIndex);
79+
writeUUID(out, invocationUid);
7780
}
7881

7982
@Override
8083
public void readData(ObjectDataInput in) throws IOException {
81-
name = in.readUTF();
8284
commitIndex = in.readLong();
85+
invocationUid = readUUID(in);
8386
}
8487

8588
@Override
@@ -96,19 +99,18 @@ public boolean equals(Object o) {
9699
if (commitIndex != that.commitIndex) {
97100
return false;
98101
}
99-
return name.equals(that.name);
102+
return invocationUid.equals(that.invocationUid);
100103
}
101104

102105
@Override
103106
public int hashCode() {
104-
int result = name.hashCode();
105-
result = 31 * result + (int) (commitIndex ^ (commitIndex >>> 32));
107+
int result = (int) (commitIndex ^ (commitIndex >>> 32));
108+
result = 31 * result + invocationUid.hashCode();
106109
return result;
107110
}
108111

109112
@Override
110113
public String toString() {
111-
return "AwaitInvocationKey{" + "name='" + name + '\'' + ", commitIndex=" + commitIndex + '}';
114+
return "AwaitInvocationKey{" + "commitIndex=" + commitIndex + ", invocationUid=" + invocationUid + '}';
112115
}
113-
114116
}

0 commit comments

Comments
 (0)