Skip to content

Commit

Permalink
RaftSemaphore operation timeout fixes
Browse files Browse the repository at this point in the history
  • Loading branch information
metanet authored and mdogan committed Feb 1, 2019
1 parent a871011 commit 0bb6683
Show file tree
Hide file tree
Showing 22 changed files with 918 additions and 107 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@
import static com.hazelcast.raft.service.util.ClientAccessor.getClient;
import static com.hazelcast.util.Preconditions.checkNotNegative;
import static com.hazelcast.util.Preconditions.checkPositive;
import static com.hazelcast.util.ThreadUtil.getThreadId;
import static com.hazelcast.util.UuidUtil.newUnsecureUUID;
import static java.lang.Math.max;

Expand Down Expand Up @@ -118,12 +119,14 @@ public void acquire() {
@Override
public void acquire(int permits) {
checkPositive(permits, "Permits must be positive!");
long threadId = getThreadId();
UUID invocationUid = newUnsecureUUID();
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 4
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 5
+ Bits.INT_SIZE_IN_BYTES;
for (;;) {
long sessionId = acquireSession(permits);
ClientMessage msg = prepareClientMessage(groupId, name, sessionId, dataSize, ACQUIRE_PERMITS_TYPE);
msg.set(threadId);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.set(permits);
Expand Down Expand Up @@ -158,14 +161,16 @@ public boolean tryAcquire(long timeout, TimeUnit unit) {
public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
checkPositive(permits, "Permits must be positive!");
long timeoutMs = max(0, unit.toMillis(timeout));
long threadId = getThreadId();
UUID invocationUid = newUnsecureUUID();
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 4
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 5
+ Bits.INT_SIZE_IN_BYTES;
long start;
for (;;) {
start = Clock.currentTimeMillis();
long sessionId = acquireSession(permits);
ClientMessage msg = prepareClientMessage(groupId, name, sessionId, dataSize, ACQUIRE_PERMITS_TYPE);
msg.set(threadId);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.set(permits);
Expand Down Expand Up @@ -202,10 +207,12 @@ public void release(int permits) {
throw new IllegalStateException("No valid session!");
}

long threadId = getThreadId();
UUID invocationUid = newUnsecureUUID();
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) +
Bits.LONG_SIZE_IN_BYTES * 3 + Bits.INT_SIZE_IN_BYTES;
Bits.LONG_SIZE_IN_BYTES * 4 + Bits.INT_SIZE_IN_BYTES;
ClientMessage msg = prepareClientMessage(groupId, name, sessionId, dataSize, RELEASE_PERMITS_TYPE);
msg.set(threadId);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.set(permits);
Expand All @@ -232,11 +239,13 @@ public int availablePermits() {

@Override
public int drainPermits() {
long threadId = getThreadId();
UUID invocationUid = newUnsecureUUID();
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 3;
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 4;
for (;;) {
long sessionId = acquireSession(DRAIN_SESSION_ACQ_COUNT);
ClientMessage msg = prepareClientMessage(groupId, name, sessionId, dataSize, DRAIN_PERMITS_TYPE);
msg.set(threadId);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.updateFrameLength();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -29,12 +29,14 @@
import com.hazelcast.raft.impl.RaftGroupIdImpl;
import com.hazelcast.raft.service.semaphore.RaftSemaphoreService;
import com.hazelcast.spi.InternalCompletableFuture;
import com.hazelcast.util.ConstructorFunction;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

import static com.hazelcast.client.impl.protocol.util.ParameterUtil.calculateDataSize;
import static com.hazelcast.raft.impl.RaftGroupIdImpl.dataSize;
import static com.hazelcast.raft.service.atomiclong.client.AtomicLongMessageTaskFactoryProvider.ADD_AND_GET_TYPE;
import static com.hazelcast.raft.service.semaphore.client.SemaphoreMessageTaskFactoryProvider.ACQUIRE_PERMITS_TYPE;
import static com.hazelcast.raft.service.semaphore.client.SemaphoreMessageTaskFactoryProvider.AVAILABLE_PERMITS_TYPE;
import static com.hazelcast.raft.service.semaphore.client.SemaphoreMessageTaskFactoryProvider.CHANGE_PERMITS_TYPE;
Expand All @@ -43,6 +45,7 @@
import static com.hazelcast.raft.service.semaphore.client.SemaphoreMessageTaskFactoryProvider.DRAIN_PERMITS_TYPE;
import static com.hazelcast.raft.service.semaphore.client.SemaphoreMessageTaskFactoryProvider.INIT_SEMAPHORE_TYPE;
import static com.hazelcast.raft.service.semaphore.client.SemaphoreMessageTaskFactoryProvider.RELEASE_PERMITS_TYPE;
import static com.hazelcast.raft.service.semaphore.proxy.GloballyUniqueThreadIdUtil.getGlobalThreadId;
import static com.hazelcast.raft.service.session.AbstractSessionManager.NO_SESSION_ID;
import static com.hazelcast.raft.service.util.ClientAccessor.getClient;
import static com.hazelcast.util.Preconditions.checkNotNegative;
Expand All @@ -56,6 +59,7 @@
public class RaftSessionlessSemaphoreProxy implements ISemaphore {

private static final ClientMessageDecoder INT_RESPONSE_DECODER = new IntResponseDecoder();
private static final ClientMessageDecoder LONG_RESPONSE_DECODER = new LongResponseDecoder();
private static final ClientMessageDecoder BOOLEAN_RESPONSE_DECODER = new BooleanResponseDecoder();

public static ISemaphore create(HazelcastInstance instance, String name) {
Expand Down Expand Up @@ -85,11 +89,32 @@ public RaftGroupId decodeClientMessage(ClientMessage msg) {
private final HazelcastClientInstanceImpl client;
private final RaftGroupId groupId;
private final String name;
private final ConstructorFunction<RaftGroupId, Long> globallyUniqueThreadIdCtor;

private RaftSessionlessSemaphoreProxy(HazelcastInstance instance, RaftGroupId groupId, String name) {
private RaftSessionlessSemaphoreProxy(HazelcastInstance instance, RaftGroupId groupId, final String name) {
this.client = getClient(instance);
this.groupId = groupId;
this.name = name;
this.globallyUniqueThreadIdCtor = new ConstructorFunction<RaftGroupId, Long>() {
@Override
public Long createNew(RaftGroupId groupId) {
int dataSize = ClientMessage.HEADER_SIZE
+ RaftGroupIdImpl.dataSize(groupId) + calculateDataSize(name)
+ Bits.LONG_SIZE_IN_BYTES;

ClientMessage msg = ClientMessage.createForEncode(dataSize);
msg.setMessageType(ADD_AND_GET_TYPE);
msg.setRetryable(false);
msg.setOperationName("");
RaftGroupIdImpl.writeTo(groupId, msg);
msg.set(name);
msg.set(1L);
msg.updateFrameLength();

ClientInvocationFuture future = new ClientInvocation(client, msg, getName()).invoke();
return new ClientDelegatingFuture<Long>(future, client.getSerializationService(), LONG_RESPONSE_DECODER).join();
}
};
}

@Override
Expand All @@ -115,10 +140,12 @@ public void acquire() {
public void acquire(int permits) {
checkPositive(permits, "Permits must be positive!");

long globalThreadId = getGlobalThreadId(groupId, globallyUniqueThreadIdCtor);
UUID invocationUid = newUnsecureUUID();
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 4
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 5
+ Bits.INT_SIZE_IN_BYTES;
ClientMessage msg = prepareClientMessage(groupId, name, dataSize, ACQUIRE_PERMITS_TYPE);
msg.set(globalThreadId);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.set(permits);
Expand Down Expand Up @@ -147,11 +174,13 @@ public boolean tryAcquire(long timeout, TimeUnit unit) {
public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
checkPositive(permits, "Permits must be positive!");

long globalThreadId = getGlobalThreadId(groupId, globallyUniqueThreadIdCtor);
UUID invocationUid = newUnsecureUUID();
long timeoutMs = max(0, unit.toMillis(timeout));
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 4
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 5
+ Bits.INT_SIZE_IN_BYTES;
ClientMessage msg = prepareClientMessage(groupId, name, dataSize, ACQUIRE_PERMITS_TYPE);
msg.set(globalThreadId);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.set(permits);
Expand All @@ -171,10 +200,12 @@ public void release() {
public void release(int permits) {
checkPositive(permits, "Permits must be positive!");

long globalThreadId = getGlobalThreadId(groupId, globallyUniqueThreadIdCtor);
UUID invocationUid = newUnsecureUUID();
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 3
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 4
+ Bits.INT_SIZE_IN_BYTES;
ClientMessage msg = prepareClientMessage(groupId, name, dataSize, RELEASE_PERMITS_TYPE);
msg.set(globalThreadId);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.set(permits);
Expand All @@ -195,9 +226,11 @@ public int availablePermits() {

@Override
public int drainPermits() {
long globalThreadId = getGlobalThreadId(groupId, globallyUniqueThreadIdCtor);
UUID invocationUid = newUnsecureUUID();
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 3;
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) + Bits.LONG_SIZE_IN_BYTES * 4;
ClientMessage msg = prepareClientMessage(groupId, name, dataSize, DRAIN_PERMITS_TYPE);
msg.set(globalThreadId);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.updateFrameLength();
Expand Down Expand Up @@ -294,6 +327,13 @@ public Integer decodeClientMessage(ClientMessage msg) {
}
}

private static class LongResponseDecoder implements ClientMessageDecoder {
@Override
public Long decodeClientMessage(ClientMessage msg) {
return msg.getLong();
}
}

private static class BooleanResponseDecoder implements ClientMessageDecoder {
@Override
public Boolean decodeClientMessage(ClientMessage msg) {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import com.hazelcast.core.ISemaphore;
import com.hazelcast.raft.RaftGroupId;
import com.hazelcast.raft.impl.session.SessionExpiredException;
import com.hazelcast.raft.service.lock.exception.LockRequestCancelledException;
import com.hazelcast.raft.service.exception.WaitKeyCancelledException;
import com.hazelcast.raft.service.semaphore.RaftSessionAwareSemaphoreBasicTest;
import com.hazelcast.raft.service.session.AbstractSessionManager;
import com.hazelcast.raft.service.session.SessionManagerProvider;
Expand Down Expand Up @@ -34,7 +34,7 @@ protected HazelcastInstance[] createInstances() {
TestHazelcastFactory f = (TestHazelcastFactory) factory;
semaphoreInstance = f.newHazelcastClient();
SessionExpiredException.register(getClient(semaphoreInstance).getClientExceptionFactory());
LockRequestCancelledException.register(getClient(semaphoreInstance).getClientExceptionFactory());
WaitKeyCancelledException.register(getClient(semaphoreInstance).getClientExceptionFactory());
return instances;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import com.hazelcast.core.ISemaphore;
import com.hazelcast.raft.RaftGroupId;
import com.hazelcast.raft.impl.session.SessionExpiredException;
import com.hazelcast.raft.service.lock.exception.LockRequestCancelledException;
import com.hazelcast.raft.service.exception.WaitKeyCancelledException;
import com.hazelcast.raft.service.semaphore.RaftSessionlessSemaphoreBasicTest;
import com.hazelcast.test.HazelcastSerialClassRunner;
import com.hazelcast.test.TestHazelcastInstanceFactory;
Expand Down Expand Up @@ -34,7 +34,7 @@ protected HazelcastInstance[] createInstances() {
TestHazelcastFactory f = (TestHazelcastFactory) factory;
client = f.newHazelcastClient();
SessionExpiredException.register(getClient(client).getClientExceptionFactory());
LockRequestCancelledException.register(getClient(client).getClientExceptionFactory());
WaitKeyCancelledException.register(getClient(client).getClientExceptionFactory());
return instances;
}

Expand Down

0 comments on commit 0bb6683

Please sign in to comment.