Skip to content
Permalink
Browse files

Raft lock & semaphore proxy clean up

* Make fenced lock proxy stateless
* Do not retry on OperationTimeoutException
* Minor cleanups & refactorings
  • Loading branch information...
metanet authored and mdogan committed Aug 1, 2018
1 parent 1fb9eaa commit e70df3cc3f4fedb05fa58eaa7c4b3bcf47b902d2
Showing with 129 additions and 247 deletions.
  1. +5 −5 hazelcast-raft-client/src/main/java/com/hazelcast/raft/service/lock/client/RaftFencedLockProxy.java
  2. +9 −9 hazelcast-raft-client/src/main/java/com/hazelcast/raft/service/lock/client/RaftLockProxy.java
  3. +20 −31 ...ent/src/main/java/com/hazelcast/raft/service/semaphore/client/RaftSessionAwareSemaphoreProxy.java
  4. +3 −3 ...ient/src/main/java/com/hazelcast/raft/service/semaphore/client/RaftSessionlessSemaphoreProxy.java
  5. +3 −1 hazelcast-raft-dataservices/src/main/java/com/hazelcast/raft/service/lock/LockRegistry.java
  6. +2 −2 hazelcast-raft-dataservices/src/main/java/com/hazelcast/raft/service/lock/RaftLockService.java
  7. +1 −1 ...ft-dataservices/src/main/java/com/hazelcast/raft/service/lock/client/GetLockFenceMessageTask.java
  8. +4 −8 ...ast-raft-dataservices/src/main/java/com/hazelcast/raft/service/lock/operation/GetLockCountOp.java
  9. +17 −2 ...ast-raft-dataservices/src/main/java/com/hazelcast/raft/service/lock/operation/GetLockFenceOp.java
  10. +28 −136 ...dataservices/src/main/java/com/hazelcast/raft/service/lock/proxy/AbstractRaftFencedLockProxy.java
  11. +4 −4 ...st-raft-dataservices/src/main/java/com/hazelcast/raft/service/lock/proxy/RaftFencedLockProxy.java
  12. +14 −12 hazelcast-raft-dataservices/src/main/java/com/hazelcast/raft/service/lock/proxy/RaftLockProxy.java
  13. +12 −27 ...ices/src/main/java/com/hazelcast/raft/service/semaphore/proxy/RaftSessionAwareSemaphoreProxy.java
  14. +1 −1 ...t-raft-dataservices/src/test/java/com/hazelcast/raft/service/lock/RaftFencedLockAdvancedTest.java
  15. +6 −5 ...cast-raft-dataservices/src/test/java/com/hazelcast/raft/service/lock/RaftFencedLockBasicTest.java
@@ -80,7 +80,7 @@ public RaftGroupId decodeClientMessage(ClientMessage msg) {

private final HazelcastClientInstanceImpl client;

public RaftFencedLockProxy(HazelcastInstance instance, RaftGroupId groupId, String name) {
private RaftFencedLockProxy(HazelcastInstance instance, RaftGroupId groupId, String name) {
super(SessionManagerProvider.get(getClient(instance)), groupId, name);
this.client = getClient(instance);
}
@@ -111,14 +111,14 @@ public RaftFencedLockProxy(HazelcastInstance instance, RaftGroupId groupId, Stri
}

@Override
protected InternalCompletableFuture<Long> doGetLockFence(RaftGroupId groupId, String name) {
ClientMessage msg = encodeRequest(LOCK_FENCE_TYPE, groupId, name, -1, -1);
protected InternalCompletableFuture<Long> doGetLockFence(RaftGroupId groupId, String name, long sessionId, long threadId) {
ClientMessage msg = encodeRequest(LOCK_FENCE_TYPE, groupId, name, sessionId, threadId);
return invoke(client, name, msg, LONG_RESPONSE_DECODER);
}

@Override
protected InternalCompletableFuture<Integer> doGetLockCount(RaftGroupId groupId, String name) {
ClientMessage msg = encodeRequest(LOCK_COUNT_TYPE, groupId, name, -1, -1);
protected InternalCompletableFuture<Integer> doGetLockCount(RaftGroupId groupId, String name, long sessionId, long threadId) {
ClientMessage msg = encodeRequest(LOCK_COUNT_TYPE, groupId, name, sessionId, threadId);
return invoke(client, name, msg, INT_RESPONSE_DECODER);
}

@@ -41,6 +41,7 @@

import static com.hazelcast.client.impl.protocol.util.ParameterUtil.calculateDataSize;
import static com.hazelcast.raft.impl.RaftGroupIdImpl.dataSize;
import static com.hazelcast.raft.service.lock.RaftLockService.INVALID_FENCE;
import static com.hazelcast.raft.service.lock.client.LockMessageTaskFactoryProvider.CREATE_TYPE;
import static com.hazelcast.raft.service.lock.client.LockMessageTaskFactoryProvider.DESTROY_TYPE;
import static com.hazelcast.raft.service.lock.client.LockMessageTaskFactoryProvider.FORCE_UNLOCK_TYPE;
@@ -49,6 +50,7 @@
import static com.hazelcast.raft.service.lock.client.LockMessageTaskFactoryProvider.LOCK_TYPE;
import static com.hazelcast.raft.service.lock.client.LockMessageTaskFactoryProvider.TRY_LOCK_TYPE;
import static com.hazelcast.raft.service.lock.client.LockMessageTaskFactoryProvider.UNLOCK_TYPE;
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.ThreadUtil.getThreadId;

@@ -102,9 +104,8 @@ public void lock() {
for (;;) {
long sessionId = acquireSession();
ClientMessage msg = encodeRequest(LOCK_TYPE, groupId, name, sessionId, getThreadId(), invUid);
InternalCompletableFuture<Object> future = invoke(client, name, msg, LONG_RESPONSE_DECODER);
try {
future.join();
invoke(client, name, msg, LONG_RESPONSE_DECODER).join();
break;
} catch (SessionExpiredException e) {
invalidateSession(sessionId);
@@ -124,9 +125,9 @@ public boolean tryLock(long time, TimeUnit unit) {
for (;;) {
long sessionId = acquireSession();
ClientMessage msg = encodeRequest(TRY_LOCK_TYPE, groupId, name, sessionId, getThreadId(), invUid, timeoutMs);
InternalCompletableFuture<Long> future = invoke(client, name, msg, LONG_RESPONSE_DECODER);
try {
boolean locked = future.join() > RaftLockService.INVALID_FENCE;
InternalCompletableFuture<Long> future = invoke(client, name, msg, LONG_RESPONSE_DECODER);
boolean locked = (future.join() != INVALID_FENCE);
if (!locked) {
releaseSession(sessionId);
}
@@ -140,14 +141,13 @@ public boolean tryLock(long time, TimeUnit unit) {
@Override
public void unlock() {
long sessionId = getSession();
if (sessionId < 0) {
if (sessionId == NO_SESSION_ID) {
throw new IllegalMonitorStateException();
}
UUID invUid = UuidUtil.newUnsecureUUID();
ClientMessage msg = encodeRequest(UNLOCK_TYPE, groupId, name, sessionId, getThreadId(), invUid);
InternalCompletableFuture<Object> future = invoke(client, name, msg, BOOLEAN_RESPONSE_DECODER);
try {
future.join();
invoke(client, name, msg, BOOLEAN_RESPONSE_DECODER).join();
} catch (SessionExpiredException e) {
throw new IllegalMonitorStateException("Current thread is not owner of the lock!");
} finally {
@@ -163,7 +163,7 @@ public boolean isLocked() {
@Override
public boolean isLockedByCurrentThread() {
long sessionId = getSession();
if (sessionId < 0) {
if (sessionId == NO_SESSION_ID) {
return false;
}
ClientMessage msg = encodeRequest(LOCK_COUNT_TYPE, groupId, name, sessionId, getThreadId());
@@ -173,7 +173,7 @@ public boolean isLockedByCurrentThread() {

@Override
public int getLockCount() {
ClientMessage msg = encodeRequest(LOCK_COUNT_TYPE, groupId, name, -1, -1);
ClientMessage msg = encodeRequest(LOCK_COUNT_TYPE, groupId, name, NO_SESSION_ID, 0);
InternalCompletableFuture<Integer> future = invoke(client, name, msg, INT_RESPONSE_DECODER);
return future.join();
}
@@ -24,7 +24,6 @@
import com.hazelcast.client.util.ClientDelegatingFuture;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ISemaphore;
import com.hazelcast.core.OperationTimeoutException;
import com.hazelcast.nio.Bits;
import com.hazelcast.raft.RaftGroupId;
import com.hazelcast.raft.impl.RaftGroupIdImpl;
@@ -61,8 +60,8 @@
*/
public class RaftSessionAwareSemaphoreProxy extends SessionAwareProxy implements ISemaphore {

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

public static ISemaphore create(HazelcastInstance instance, String name) {
int dataSize = ClientMessage.HEADER_SIZE + calculateDataSize(name);
@@ -91,7 +90,7 @@ public RaftGroupId decodeClientMessage(ClientMessage msg) {
private final HazelcastClientInstanceImpl client;
private final String name;

public RaftSessionAwareSemaphoreProxy(HazelcastInstance instance, RaftGroupId groupId, String name) {
private RaftSessionAwareSemaphoreProxy(HazelcastInstance instance, RaftGroupId groupId, String name) {
super(SessionManagerProvider.get(getClient(instance)), groupId);
this.client = getClient(instance);
this.name = name;
@@ -131,12 +130,9 @@ public void acquire(int permits) {
msg.set(-1L);
msg.updateFrameLength();

InternalCompletableFuture<Object> future = invoke(msg, BOOLEAN_RESPONSE_DECODER);
try {
future.join();
invoke(msg, BOOLEAN_RESPONSE_DECODER).join();
return;
} catch (OperationTimeoutException ignored) {
// I can retry safely because my retry would be idempotent...
} catch (SessionExpiredException e) {
invalidateSession(sessionId);
}
@@ -176,17 +172,19 @@ public boolean tryAcquire(int permits, long timeout, TimeUnit unit) {
msg.set(timeoutMs);
msg.updateFrameLength();

InternalCompletableFuture<Boolean> future = invoke(msg, BOOLEAN_RESPONSE_DECODER);
try {
InternalCompletableFuture<Boolean> future = invoke(msg, BOOLEAN_RESPONSE_DECODER);
boolean acquired = future.join();
if (!acquired) {
releaseSession(sessionId, permits);
}
return acquired;
} catch (OperationTimeoutException e) {
timeoutMs = Math.max(0, (timeoutMs - (Clock.currentTimeMillis() - start)));
} catch (SessionExpiredException e) {
invalidateSession(sessionId);
timeoutMs -= (Clock.currentTimeMillis() - start);
if (timeoutMs <= 0) {
return false;
}
}
}
}
@@ -200,30 +198,23 @@ public void release() {
public void release(int permits) {
checkPositive(permits, "Permits must be positive!");
long sessionId = getSession();
if (sessionId < 0) {
if (sessionId == NO_SESSION_ID) {
throw new IllegalStateException("No valid session!");
}

UUID invocationUid = newUnsecureUUID();
int dataSize = ClientMessage.HEADER_SIZE + dataSize(groupId) + calculateDataSize(name) +
Bits.LONG_SIZE_IN_BYTES * 3 + Bits.INT_SIZE_IN_BYTES;
ClientMessage msg = prepareClientMessage(groupId, name, sessionId, dataSize, RELEASE_PERMITS_TYPE);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.set(permits);
msg.updateFrameLength();
try {
for (;;) {
ClientMessage msg = prepareClientMessage(groupId, name, sessionId, dataSize, RELEASE_PERMITS_TYPE);
msg.set(invocationUid.getLeastSignificantBits());
msg.set(invocationUid.getMostSignificantBits());
msg.set(permits);
msg.updateFrameLength();
try {
invoke(msg, BOOLEAN_RESPONSE_DECODER).join();
return;
} catch (OperationTimeoutException ignored) {
// I can retry safely because my retry would be idempotent...
} catch (SessionExpiredException e) {
invalidateSession(sessionId);
throw e;
}
}
invoke(msg, BOOLEAN_RESPONSE_DECODER).join();
} catch (SessionExpiredException e) {
invalidateSession(sessionId);
throw e;
} finally {
releaseSession(sessionId, permits);
}
@@ -250,13 +241,11 @@ public int drainPermits() {
msg.set(invocationUid.getMostSignificantBits());
msg.updateFrameLength();

InternalCompletableFuture<Integer> future = invoke(msg, INT_RESPONSE_DECODER);
try {
InternalCompletableFuture<Integer> future = invoke(msg, INT_RESPONSE_DECODER);
int count = future.join();
releaseSession(sessionId, DRAIN_SESSION_ACQ_COUNT - count);
return count;
} catch (OperationTimeoutException ignored) {
// I can retry safely because my retry would be idempotent...
} catch (SessionExpiredException e) {
invalidateSession(sessionId);
}
@@ -55,8 +55,8 @@
*/
public class RaftSessionlessSemaphoreProxy implements ISemaphore {

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

public static ISemaphore create(HazelcastInstance instance, String name) {
int dataSize = ClientMessage.HEADER_SIZE + calculateDataSize(name);
@@ -86,7 +86,7 @@ public RaftGroupId decodeClientMessage(ClientMessage msg) {
private final RaftGroupId groupId;
private final String name;

public RaftSessionlessSemaphoreProxy(HazelcastInstance instance, RaftGroupId groupId, String name) {
private RaftSessionlessSemaphoreProxy(HazelcastInstance instance, RaftGroupId groupId, String name) {
this.client = getClient(instance);
this.groupId = groupId;
this.name = name;
@@ -99,7 +99,7 @@ int getLockCount(String name, LockEndpoint endpoint) {
return lock.lockCount();
}

long getLockFence(String name) {
long getLockFence(String name, LockEndpoint endpoint) {
RaftLock lock = getResourceOrNull(name);
if (lock == null) {
throw new IllegalMonitorStateException();
@@ -108,6 +108,8 @@ long getLockFence(String name) {
LockInvocationKey owner = lock.owner();
if (owner == null) {
throw new IllegalMonitorStateException("Lock[" + name + "] has no owner!");
} else if (endpoint != null && !owner.endpoint().equals(endpoint)) {
throw new IllegalMonitorStateException("Lock[" + name + "] is owned by " + owner.endpoint() + "!");
}

return owner.commitIndex();
@@ -144,11 +144,11 @@ public int getLockCount(RaftGroupId groupId, String name, LockEndpoint endpoint)
return registry != null ? registry.getLockCount(name, endpoint) : 0;
}

public long getLockFence(RaftGroupId groupId, String name) {
public long getLockFence(RaftGroupId groupId, String name, LockEndpoint endpoint) {
checkNotNull(groupId);
checkNotNull(name);

return getLockRegistryOrFail(groupId, name).getLockFence(name);
return getLockRegistryOrFail(groupId, name).getLockFence(name, endpoint);
}

private LockRegistry getLockRegistryOrFail(RaftGroupId groupId, String name) {
@@ -36,7 +36,7 @@
@Override
protected void processMessage() {
RaftInvocationManager invocationManager = getRaftInvocationManager();
invocationManager.invoke(groupId, new GetLockFenceOp(name)).andThen(this);
invocationManager.invoke(groupId, new GetLockFenceOp(name, sessionId, threadId)).andThen(this);
}

@Override
@@ -27,24 +27,20 @@

import java.io.IOException;

import static com.hazelcast.raft.service.session.AbstractSessionManager.NO_SESSION_ID;

/**
* TODO: Javadoc Pending...
*/
public class GetLockCountOp extends RaftOp implements IdentifiedDataSerializable {

private static int NO_SESSION = -1;

private String name;
private long sessionId = NO_SESSION;
private long sessionId;
private long threadId;

public GetLockCountOp() {
}

public GetLockCountOp(String name) {
this.name = name;
}

public GetLockCountOp(String name, long sessionId, long threadId) {
this.name = name;
this.sessionId = sessionId;
@@ -54,7 +50,7 @@ public GetLockCountOp(String name, long sessionId, long threadId) {
@Override
public Object run(RaftGroupId groupId, long commitIndex) {
RaftLockService service = getService();
LockEndpoint endpoint = (sessionId != NO_SESSION) ? new LockEndpoint(sessionId, threadId) : null;
LockEndpoint endpoint = (sessionId != NO_SESSION_ID) ? new LockEndpoint(sessionId, threadId) : null;

return service.getLockCount(groupId, name, endpoint);
}
@@ -21,29 +21,38 @@
import com.hazelcast.nio.serialization.IdentifiedDataSerializable;
import com.hazelcast.raft.RaftGroupId;
import com.hazelcast.raft.impl.RaftOp;
import com.hazelcast.raft.service.lock.LockEndpoint;
import com.hazelcast.raft.service.lock.RaftLockDataSerializerHook;
import com.hazelcast.raft.service.lock.RaftLockService;

import java.io.IOException;

import static com.hazelcast.raft.service.session.AbstractSessionManager.NO_SESSION_ID;

/**
* TODO: Javadoc Pending...
*/
public class GetLockFenceOp extends RaftOp implements IdentifiedDataSerializable {

private String name;
private long sessionId;
private long threadId;

public GetLockFenceOp() {
}

public GetLockFenceOp(String name) {
public GetLockFenceOp(String name, long sessionId, long threadId) {
this.name = name;
this.sessionId = sessionId;
this.threadId = threadId;
}

@Override
public Object run(RaftGroupId groupId, long commitIndex) {
RaftLockService service = getService();
return service.getLockFence(groupId, name);
LockEndpoint endpoint = (sessionId != NO_SESSION_ID) ? new LockEndpoint(sessionId, threadId) : null;

return service.getLockFence(groupId, name, endpoint);
}

@Override
@@ -64,15 +73,21 @@ public int getId() {
@Override
public void writeData(ObjectDataOutput out) throws IOException {
out.writeUTF(name);
out.writeLong(sessionId);
out.writeLong(threadId);
}

@Override
public void readData(ObjectDataInput in) throws IOException {
name = in.readUTF();
sessionId = in.readLong();
threadId = in.readLong();
}

@Override
protected void toString(StringBuilder sb) {
sb.append(", name=").append(name);
sb.append(", sessionId=").append(sessionId);
sb.append(", threadId=").append(threadId);
}
}

0 comments on commit e70df3c

Please sign in to comment.
You can’t perform that action at this time.