| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| |
| package org.apache.geode.distributed.internal.locks; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.IOException; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelCriterion; |
| import org.apache.geode.CancelException; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.InternalGemFireError; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.annotations.internal.MutableForTesting; |
| import org.apache.geode.distributed.LockServiceDestroyedException; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.DistributionMessage; |
| import org.apache.geode.distributed.internal.HighPriorityDistributionMessage; |
| import org.apache.geode.distributed.internal.MessageWithReply; |
| import org.apache.geode.distributed.internal.ProcessorKeeper21; |
| import org.apache.geode.distributed.internal.ReplyException; |
| import org.apache.geode.distributed.internal.ReplyMessage; |
| import org.apache.geode.distributed.internal.ReplyProcessor21; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| import org.apache.geode.internal.serialization.DeserializationContext; |
| import org.apache.geode.internal.serialization.SerializationContext; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| |
| /** |
| * Provides handling of remote and local lock requests. <br> |
| * A lock client sends a <code>DLockRequestMessage</code> to the lock grantor and then blocks, |
| * waiting for the reply. <br> |
| * When the lock grantor grants or times out the request, a <code>DLockResponseMessage</code> is |
| * finally sent back to the waiting client. |
| * |
| */ |
| public class DLockRequestProcessor extends ReplyProcessor21 { |
| private static final Logger logger = LogService.getLogger(); |
| |
| protected final DLockRequestMessage request; |
| |
| private final DistributionManager dm; |
| |
| protected final DLockService svc; |
| |
| private final InternalDistributedMember grantor; |
| |
| private volatile boolean gotLock = false; |
| |
| private DLockResponseMessage response; |
| |
| private final boolean disableAlerts; |
| |
| @Override |
| protected boolean processTimeout() { |
| // if this returns false then no need to log warning/severe msg as Dlock request will have |
| // timeout |
| return !disableAlerts; |
| } |
| |
| // private volatile boolean doneProcessing = false; |
| |
| // private final long grantorVersion; |
| |
| protected static ProcessorKeeper21 getKeeper() { |
| return ReplyProcessor21.keeper; |
| } |
| |
| // ------------------------------------------------------------------------- |
| // Constructors |
| // ------------------------------------------------------------------------- |
| |
| protected DLockRequestProcessor(LockGrantorId lockGrantorId, DLockService svc, Object objectName, |
| int threadId, long startTime, long leaseMillis, long waitMillis, boolean reentrant, |
| boolean tryLock, DistributionManager dm) { |
| this(lockGrantorId, svc, objectName, threadId, startTime, leaseMillis, waitMillis, reentrant, |
| tryLock, false, dm, false); |
| } |
| |
| protected DLockRequestProcessor(LockGrantorId lockGrantorId, DLockService svc, Object objectName, |
| int threadId, long startTime, long leaseMillis, long waitMillis, boolean reentrant, |
| boolean tryLock, boolean disableAlerts, DistributionManager dm) { |
| this(lockGrantorId, svc, objectName, threadId, startTime, leaseMillis, waitMillis, reentrant, |
| tryLock, disableAlerts, dm, false); |
| } |
| |
| protected DLockRequestProcessor(LockGrantorId lockGrantorId, DLockService svc, Object objectName, |
| int threadId, long startTime, long leaseMillis, long waitMillis, boolean reentrant, |
| boolean tryLock, DistributionManager dm, boolean async) { |
| this(lockGrantorId, svc, objectName, threadId, startTime, leaseMillis, waitMillis, reentrant, |
| tryLock, false, dm, false); |
| } |
| |
| protected DLockRequestProcessor(LockGrantorId lockGrantorId, DLockService svc, Object objectName, |
| int threadId, long startTime, long leaseMillis, long waitMillis, boolean reentrant, |
| boolean tryLock, boolean disableAlerts, DistributionManager dm, boolean async) { |
| super(dm, lockGrantorId.getLockGrantorMember()); |
| |
| this.svc = svc; |
| this.dm = dm; |
| grantor = lockGrantorId.getLockGrantorMember(); |
| // this.grantorVersion = grantorVersion; |
| |
| request = createRequest(); |
| Assert.assertTrue(getProcessorId() > 0); |
| |
| request.processorId = getProcessorId(); |
| request.serviceName = svc.getName(); |
| request.objectName = objectName; |
| request.threadId = threadId; |
| request.startTime = startTime; |
| request.leaseMillis = leaseMillis; |
| request.waitMillis = waitMillis; |
| request.reentrant = reentrant; |
| request.tryLock = tryLock; |
| request.grantorVersion = lockGrantorId.getLockGrantorVersion(); |
| request.grantorSerialNumber = lockGrantorId.getLockGrantorSerialNumber(); |
| request.dlsSerialNumber = svc.getSerialNumber(); |
| |
| request.setRecipient(grantor); |
| this.disableAlerts = disableAlerts; |
| } |
| |
| protected DLockRequestMessage createRequest() { |
| return new DLockRequestMessage(); |
| } |
| |
| protected CancelCriterion getCancelCriterion(DistributionManager ignoreDM) { |
| return svc.getCancelCriterion(); |
| } |
| |
| boolean repliedDestroyed() { |
| if (response == null) { |
| return false; |
| } |
| return response.responseCode == DLockResponseMessage.DESTROYED; |
| } |
| |
| boolean repliedNotHolder() { |
| if (response == null) { |
| return false; |
| } |
| return response.responseCode == DLockResponseMessage.NOT_HOLDER; |
| } |
| |
| boolean repliedNotGrantor() { |
| if (response == null) { |
| return false; |
| } |
| return response.responseCode == DLockResponseMessage.NOT_GRANTOR; |
| } |
| |
| boolean hadNoResponse() { |
| return response == null; |
| } |
| |
| boolean tryLockFailed() { |
| if (response == null) { |
| return false; |
| } |
| return response.responseCode == DLockResponseMessage.TRY_LOCK_FAILED; |
| } |
| |
| String getResponseCodeString() { |
| if (response == null) { |
| return null; |
| } |
| return DLockResponseMessage.responseCodeToString(response.responseCode); |
| } |
| |
| public DLockResponseMessage getResponse() { |
| return response; |
| } |
| |
| long getLeaseExpireTime() { |
| return response.leaseExpireTime; |
| } |
| |
| protected boolean requestLock(boolean interruptible, int lockId) throws InterruptedException { |
| final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE); |
| |
| Assert.assertTrue(lockId > -1, "lockId is < 0: " + this); |
| request.lockId = lockId; |
| |
| // local grantor... don't use messaging... fake it |
| if (isLockGrantor()) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "DLockRequestProcessor processing lock request directly"); |
| } |
| request.setSender(dm.getDistributionManagerId()); |
| |
| // calls processor (this) process... |
| request.processLocally(dm); |
| } |
| |
| // remote grantor... use messaging |
| else { |
| // send the message... |
| dm.putOutgoing(request); |
| } |
| |
| if (interruptible) { |
| try { |
| waitForReplies(); |
| } catch (ReplyException ex) { |
| if (ex.getCause() instanceof InterruptedException) { |
| throw (InterruptedException) ex.getCause(); |
| } |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "DLockRequestProcessor caught ReplyException", ex); |
| } |
| return false; |
| } |
| } else { // not interruptible |
| try { |
| waitForRepliesUninterruptibly(); |
| } catch (ReplyException ex) { |
| if (ex.getCause() instanceof InterruptedException) { |
| throw (InterruptedException) ex.getCause(); |
| } |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "DLockRequestProcessor caught ReplyException", ex); |
| } |
| return false; |
| } |
| } |
| |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "DLockRequestProcessor {} for {}", |
| (gotLock ? "got lock" : "failed to get lock"), request); |
| } |
| return gotLock; |
| } |
| |
| @Override |
| protected boolean allowReplyFromSender() { |
| return true; |
| } |
| |
| private boolean isLockGrantor() { |
| return dm.getDistributionManagerId().equals(grantor); |
| } |
| |
| Object getKeyIfFailed() { |
| if (gotLock || response == null) { |
| return null; |
| } |
| return response.keyIfFailed; |
| } |
| |
| protected boolean gotLock() { |
| return gotLock; |
| } |
| |
| @Override |
| public void process(DistributionMessage msg) { |
| final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE); |
| try { |
| Assert.assertTrue(msg instanceof DLockResponseMessage, |
| "DLockRequestProcessor is unable to process message of type " + msg.getClass()); |
| |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "Processing DLockResponseMessage: '{}'", msg); |
| } |
| final DLockResponseMessage reply = (DLockResponseMessage) msg; |
| response = reply; |
| |
| if (response.getLockId() != request.getLockId()) { |
| // Ignore this response since it was sent for a lockId that |
| // must have timed out. |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "Failed to find processor for lockId {} processor ids must have wrapped.", |
| response.getLockId()); |
| } |
| Assert.assertTrue(response.getLockId() == request.getLockId()); |
| } |
| |
| switch (reply.responseCode) { |
| case DLockResponseMessage.GRANT: |
| // grantor has granted the lock request... |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "{} has granted lock for {} in {}", |
| reply.getSender(), reply.objectName, reply.serviceName); |
| } |
| gotLock = true; |
| break; |
| case DLockResponseMessage.NOT_GRANTOR: |
| // target was not the grantor! who is the grantor?! |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "{} has responded DLockResponseMessage.NOT_GRANTOR for {}", reply.getSender(), |
| reply.serviceName); |
| } |
| break; |
| case DLockResponseMessage.DESTROYED: |
| // grantor claims we sent it a NonGrantorDestroyedMessage |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "{} has responded DLockResponseMessage.DESTROYED for {}", reply.getSender(), |
| reply.serviceName); |
| } |
| break; |
| case DLockResponseMessage.TIMEOUT: |
| // grantor told us the lock request has timed out... |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "{} has responded DLockResponseMessage.TIMEOUT for {} in {}", reply.getSender(), |
| reply.objectName, reply.serviceName); |
| } |
| break; |
| case DLockResponseMessage.SUSPENDED: |
| // grantor told us that locking has been suspended for the service... |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "{} has responded DLockResponseMessage.SUSPENDED for {} in {}", reply.getSender(), |
| reply.objectName, reply.serviceName); |
| } |
| break; |
| case DLockResponseMessage.NOT_HOLDER: |
| // tried to reenter lock but grantor says we're not the lock holder... |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "{} has responded DLockResponseMessage.NOT_HOLDER for {} in {}", reply.getSender(), |
| reply.objectName, reply.serviceName); |
| } |
| break; |
| case DLockResponseMessage.TRY_LOCK_FAILED: |
| // tried to acquire try-lock but grantor says it's held and we failed... |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "{} has responded DLockResponseMessage.TRY_LOCK_FAILED for {} in {}", |
| reply.getSender(), reply.objectName, reply.serviceName); |
| } |
| break; |
| default: |
| throw new InternalGemFireError( |
| String.format("Unknown response code %s", |
| reply.responseCode)); |
| } // switch |
| |
| } finally { |
| super.process(msg); |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "Finished processing DLockResponseMessage: '{}'", msg); |
| } |
| ((DLockResponseMessage) msg).processed = true; |
| } |
| } |
| |
| /** |
| * LockGrantorDestroyedException or LockServiceDestroyedException is an anticipated reply |
| * exception. Receiving multiple replies with this exception is normal. |
| */ |
| @Override |
| protected boolean logMultipleExceptions() { |
| return false; |
| } |
| |
| // ------------------------------------------------------------------------- |
| // DLockRequestMessage |
| // ------------------------------------------------------------------------- |
| public static class DLockRequestMessage extends HighPriorityDistributionMessage |
| implements MessageWithReply { |
| /** |
| * The id of the DLockRequestProcessor on the initiator node. This will be communicated back in |
| * the response to enable collation of the results. |
| */ |
| protected int processorId; |
| |
| /** The name of the DistributedLockService */ |
| protected String serviceName; |
| |
| /** The object name */ |
| protected Object objectName; |
| |
| /** Time when lock request was initiated */ |
| protected long startTime; |
| |
| protected long leaseMillis; |
| |
| protected long waitMillis; |
| |
| /** True if re-entering held lock */ |
| protected boolean reentrant; |
| |
| /** True if try lock and respond immediately without scheduling */ |
| protected boolean tryLock; |
| |
| /** Uniquely identifies this request for later releasing or re-entry */ |
| protected int lockId; |
| |
| /** Uniquely identifies the thread making this request */ |
| protected int threadId; |
| |
| /** version of the grantor that this request is targeted toward */ |
| protected long grantorVersion; |
| |
| /** serial number of grantor's DLS that this request is going to */ |
| protected int grantorSerialNumber; |
| |
| /** serial number of the DLockService that originated this request */ |
| protected int dlsSerialNumber; |
| |
| protected transient DLockService svc; |
| protected transient DLockGrantor grantor; |
| private transient long statStart = -1; |
| transient volatile DistributionManager receivingDM; |
| transient DLockResponseMessage response; |
| private transient RemoteThread rThread; |
| |
| /** True if we've responded to this request */ |
| private boolean responded = false; |
| |
| public DLockRequestMessage() {} |
| |
| public boolean isLocal() { |
| Assert.assertTrue(receivingDM != null); |
| return getSender().equals(receivingDM.getId()); |
| } |
| |
| public boolean isTryLock() { |
| return tryLock; |
| } |
| |
| @Override |
| public int getProcessorId() { |
| return processorId; |
| } |
| |
| public Object getObjectName() { |
| return objectName; |
| } |
| |
| public long getStartTime() { |
| return startTime; |
| } |
| |
| public long getLeaseTime() { |
| return leaseMillis; |
| } |
| |
| public long getWaitTime() { |
| return waitMillis; |
| } |
| |
| public int getLockId() { |
| return lockId; |
| } |
| |
| public int getThreadId() { |
| return threadId; |
| } |
| |
| public long getGrantorVersion() { |
| return grantorVersion; |
| } |
| |
| public int getGrantorSerialNumber() { |
| return grantorSerialNumber; |
| } |
| |
| private final transient Object rThreadLock = new Object(); |
| |
| public RemoteThread getRemoteThread() { |
| synchronized (rThreadLock) { |
| if (rThread == null) { |
| // grantor will need RemoteThread to process this request... |
| rThread = new RemoteThread(getSender(), getThreadId()); |
| } |
| return rThread; |
| } |
| } |
| |
| boolean isSuspendLockingRequest() { |
| return getObjectName().equals(DLockService.SUSPEND_LOCKING_TOKEN); |
| } |
| |
| // void setReceivingDM(DM dm) { |
| // this.receivingDM = dm; |
| // |
| // this.response = createResponse(); |
| // this.response.setProcessorId(getProcessorId()); |
| // this.response.setRecipient(getSender()); |
| // this.response.serviceName = this.serviceName; |
| // this.response.objectName = this.objectName; |
| // } |
| |
| private long startGrantWait() { |
| return DLockService.getDistributedLockStats().startGrantWait(); |
| } |
| |
| protected DLockResponseMessage createResponse() { |
| return new DLockResponseMessage(); |
| } |
| |
| /** |
| * Processes this message - invoked on the node that is the lock grantor. |
| */ |
| @Override |
| protected void process(final ClusterDistributionManager dm) { |
| boolean failed = false; |
| Throwable replyException = null; |
| try { |
| statStart = startGrantWait(); |
| svc = DLockService.getInternalServiceNamed(serviceName); |
| if (svc == null) { |
| failed = false; // basicProcess has it's own finally-block w reply |
| basicProcess(dm, false); // don't have a grantor anymore |
| } else { |
| executeBasicProcess(dm); // use executor |
| } |
| failed = false; // nothing above threw anything |
| } catch (RuntimeException e) { |
| replyException = e; |
| throw e; |
| } catch (VirtualMachineError e) { |
| SystemFailure.initiateFailure(e); |
| replyException = e; |
| throw e; |
| } catch (Error e) { |
| SystemFailure.checkFailure(); |
| replyException = e; |
| throw e; |
| } finally { |
| if (failed) { |
| // above code failed so now ensure reply is sent |
| if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) { |
| logger.trace(LogMarker.DLS_VERBOSE, "DLockRequestMessage.process failed for <{}>", |
| this); |
| } |
| response = createResponse(); |
| response.setProcessorId(getProcessorId()); |
| response.setRecipient(getSender()); |
| response.serviceName = serviceName; |
| response.objectName = objectName; |
| response.lockId = lockId; |
| respondWithException(replyException); |
| } |
| } |
| } |
| |
| /** Process locally without using messaging or executor */ |
| protected void processLocally(final DistributionManager dm) { |
| statStart = startGrantWait(); |
| svc = DLockService.getInternalServiceNamed(serviceName); |
| basicProcess(dm, true); // don't use executor |
| } |
| |
| /** |
| * Execute basicProcess inside Pooled Executor because grantor may not be initializing which |
| * will require us to wait. |
| * <p> |
| * this.svc and this.grantor must be set before calling this method. |
| */ |
| private void executeBasicProcess(final DistributionManager dm) { |
| final DLockRequestMessage msg = this; |
| dm.getExecutors().getWaitingThreadPool().execute(() -> { |
| if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) { |
| logger.trace(LogMarker.DLS_VERBOSE, "calling waitForGrantor {}", msg); |
| } |
| basicProcess(dm, true); |
| }); |
| } |
| |
| protected void basicProcess(final DistributionManager dm, final boolean waitForGrantor) { |
| final boolean isDebugEnabled_DLS = logger.isTraceEnabled(LogMarker.DLS_VERBOSE); |
| try { |
| receivingDM = dm; |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "DLockRequestMessage.basicProcess processing <{}>", |
| this); |
| } |
| response = createResponse(); |
| response.setProcessorId(getProcessorId()); |
| response.setRecipient(getSender()); |
| response.serviceName = serviceName; |
| response.objectName = objectName; |
| response.lockId = lockId; |
| |
| if (waitForGrantor && svc != null) { |
| try { |
| grantor = DLockGrantor.waitForGrantor(svc); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| grantor = null; // fail it |
| } |
| } |
| |
| if (svc == null || grantor == null) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "respondWithNotGrantor this.svc={} this.grantor={}", |
| svc, grantor); |
| } |
| respondWithNotGrantor(); |
| } |
| |
| else if (grantor.isDestroyed()) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "respondWithNotGrantor grantor was destroyed {}", |
| grantor); |
| } |
| respondWithNotGrantor(); |
| } else if (grantor.getVersionId() != grantorVersion) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "respondWithNotGrantor current version is {}; request was for {}", |
| grantor.getVersionId(), grantorVersion); |
| } |
| respondWithNotGrantor(); |
| } else if (svc.getSerialNumber() != grantorSerialNumber) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "respondWithNotGrantor current serial number is {}; request was for {}", |
| svc.getSerialNumber(), grantorSerialNumber); |
| } |
| respondWithNotGrantor(); |
| } |
| |
| // this is the grantor, so the request will be processed... |
| else { |
| svc.checkDestroyed(); |
| if (!svc.isLockGrantor()) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "respondWithNotGrantor service !isLockGrantor svc={}", svc); |
| } |
| respondWithNotGrantor(); |
| } |
| grantor.checkDestroyed(); |
| |
| // handle lock re-entry... |
| if (reentrant) { |
| long leaseExpireTime; |
| try { |
| leaseExpireTime = grantor.reenterLock(this); |
| } catch (InterruptedException e) { |
| leaseExpireTime = 0; // just fail it |
| } |
| if (leaseExpireTime == 0) { |
| respondWithNotHolder(); |
| } else { |
| respondWithGrant(leaseExpireTime); |
| } |
| } |
| |
| // queue up this request to be granted... |
| else { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "Handling lock request: <{}>", this); |
| } |
| if (grantor.isDestroyed()) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "respondWithNotGrantor grantor was destroyed grantor={}", grantor); |
| } |
| respondWithNotGrantor(); |
| } else { |
| try { |
| grantor.handleLockRequest(this); |
| } catch (InterruptedException | LockGrantorDestroyedException e) { |
| // just fail it |
| respondWithNotGrantor(); |
| } |
| } |
| } |
| } |
| } catch (LockGrantorDestroyedException e) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "LockGrantorDestroyedException respondWithNotGrantor svc={}", svc); |
| } |
| respondWithNotGrantor(); |
| } catch (LockServiceDestroyedException e) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "LockServiceDestroyedException respondWithNotGrantor svc={}", svc); |
| } |
| respondWithNotGrantor(); |
| } catch (CancelException e) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "CacheClosedException respondWithNotGrantor svc={} exception = {}", svc, e); |
| } |
| if (isLocal()) { |
| throw e; |
| } else { |
| respondWithNotGrantor(); |
| } |
| } catch (RuntimeException e) { |
| logger.warn(LogMarker.DLS_MARKER, |
| "[DLockRequestMessage.process] Caught throwable:", |
| e); |
| respondWithException(e); |
| } |
| } |
| |
| synchronized void respondWithNotGrantor() { |
| response.responseCode = DLockResponseMessage.NOT_GRANTOR; |
| sendResponse(); |
| } |
| |
| synchronized void respondWithDestroyed() { |
| response.responseCode = DLockResponseMessage.DESTROYED; |
| sendResponse(); |
| } |
| |
| private synchronized void respondWithNotHolder() { |
| response.responseCode = DLockResponseMessage.NOT_HOLDER; |
| sendResponse(); |
| } |
| |
| /** Callers must be synchronized on this */ |
| private void respondWithTimeout() { |
| if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) { |
| logger.trace(LogMarker.DLS_VERBOSE, "Request {} timed out; grantor status = {}", this, |
| grantor.displayStatus(rThread, objectName)); |
| } |
| response.responseCode = DLockResponseMessage.TIMEOUT; |
| sendResponse(); |
| } |
| |
| synchronized void respondWithTryLockFailed(Object keyIfFailed) { |
| response.responseCode = DLockResponseMessage.TRY_LOCK_FAILED; |
| response.keyIfFailed = keyIfFailed; |
| sendResponse(); |
| } |
| |
| synchronized void respondWithGrant(long leaseExpireTime) { |
| response.responseCode = DLockResponseMessage.GRANT; |
| response.leaseExpireTime = leaseExpireTime; |
| response.dlsSerialNumber = dlsSerialNumber; |
| sendResponse(); |
| } |
| |
| synchronized void respondWithException(Throwable t) { |
| try { |
| if (response.getException() == null) { |
| response.setException(new ReplyException(t)); |
| if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "While processing <{}>, got exception, returning to sender", this, |
| response.getException()); |
| } |
| } else { |
| logger.warn(LogMarker.DLS_VERBOSE, |
| String.format("More than one exception thrown in %s", |
| this), |
| t); |
| } |
| } finally { |
| sendResponse(); |
| } |
| } |
| |
| /** |
| * Return the timestamp at which this request should timeout. If it should never timeout then |
| * returns Long.MAX_VALUE. |
| */ |
| long getTimeoutTS() { |
| if (waitMillis == -1 || waitMillis == Long.MAX_VALUE) { |
| return Long.MAX_VALUE; |
| } else { |
| long result = startTime + waitMillis; |
| if (result < startTime) { |
| result = Long.MAX_VALUE; |
| } |
| return result; |
| } |
| } |
| |
| synchronized boolean checkForTimeout() { |
| if (waitMillis == -1 || waitMillis == Long.MAX_VALUE) { |
| return false; |
| } |
| if (tryLock) { |
| return false; |
| } |
| long now = DLockService.getLockTimeStamp(receivingDM); |
| if (now < startTime) { |
| now = startTime; |
| } |
| if (waitMillis + startTime - now <= 0) { |
| if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "DLockRequestProcessor request timed out: waitMillis={} now={} startTime={}", |
| waitMillis, now, startTime); |
| } |
| respondWithTimeout(); |
| return true; |
| } |
| return false; |
| } |
| |
| void endGrantWaitStatistic() { |
| if (statStart == -1) { |
| return; // failed to start the stat |
| } |
| DistributedLockStats stats = DLockService.getDistributedLockStats(); |
| switch (response.responseCode) { |
| case DLockResponseMessage.GRANT: |
| stats.endGrantWait(statStart); |
| break; |
| case DLockResponseMessage.NOT_GRANTOR: |
| stats.endGrantWaitNotGrantor(statStart); |
| break; |
| case DLockResponseMessage.TIMEOUT: |
| stats.endGrantWaitTimeout(statStart); |
| break; |
| case DLockResponseMessage.SUSPENDED: |
| stats.endGrantWaitSuspended(statStart); |
| break; |
| case DLockResponseMessage.NOT_HOLDER: |
| stats.endGrantWaitNotHolder(statStart); |
| break; |
| case DLockResponseMessage.TRY_LOCK_FAILED: |
| stats.endGrantWaitFailed(statStart); |
| break; |
| case DLockResponseMessage.DESTROYED: |
| stats.endGrantWaitDestroyed(statStart); |
| break; |
| default: |
| Assert.assertTrue(false, "Unknown responseCode: " + response.responseCode); |
| break; |
| } |
| } |
| |
| /** Callers must be synchronized on this */ |
| void sendResponse() { |
| try { |
| if (responded) { |
| return; |
| } |
| InternalDistributedMember myId = receivingDM.getDistributionManagerId(); |
| |
| // local... don't actually use messaging |
| if (getSender().equals(myId)) { |
| if (debugReleaseOrphanedGrant()) { |
| waitToProcessDLockResponse(receivingDM); |
| } |
| ReplyProcessor21 processor = getReplyProcessor(); |
| if (processor == null) { |
| // lock request was probably interrupted so we need to release it... |
| logger.warn(LogMarker.DLS_MARKER, |
| "Failed to find processor for {}", |
| response); |
| if (response.responseCode == DLockResponseMessage.GRANT) { |
| logger.info(LogMarker.DLS_MARKER, |
| "Releasing local orphaned grant for {}.", |
| this); |
| try { |
| grantor.releaseIfLocked(objectName, getSender(), lockId); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.warn("Interrupted while releasing grant.", e); |
| } |
| logger.info("Handled local orphaned grant."); |
| } |
| endGrantWaitStatistic(); |
| return; |
| } |
| response.setSender(getSender()); |
| endGrantWaitStatistic(); |
| processor.process(response); |
| } |
| |
| // remote... use messaging |
| else { |
| DLockResponseMessage responseMessage = response; |
| executeGrantToRemote(responseMessage); |
| } |
| } finally { |
| responded = true; |
| } |
| } |
| |
| ReplyProcessor21 getReplyProcessor() { |
| return ReplyProcessor21.getProcessor(processorId); |
| } |
| |
| void executeGrantToRemote(DLockResponseMessage responseMessage) { |
| if (!Thread.currentThread().getName().startsWith("Pooled Waiting Message Processor")) { |
| receivingDM.getExecutors().getWaitingThreadPool() |
| .execute(() -> grantToRemote(responseMessage)); |
| } else { |
| grantToRemote(responseMessage); |
| } |
| } |
| |
| void grantToRemote(DLockResponseMessage responseMessage) { |
| receivingDM.putOutgoing(responseMessage); |
| endGrantWaitStatistic(); |
| } |
| |
| synchronized void handleDepartureOfSender() { |
| try { |
| if (receivingDM.getDistributionManagerIds().contains(sender)) { |
| // sender must have sent us a NonGrantorDestroyedMessage |
| // still need to send a reply to make the thread stop waiting |
| respondWithDestroyed(); |
| } |
| } finally { |
| if (!responded) { |
| endGrantWaitStatistic(); |
| responded = true; |
| } |
| } |
| } |
| |
| synchronized boolean responded() { |
| return responded; |
| } |
| |
| /** Callers must be synchronized on this */ |
| boolean respondedNoSync() { |
| return responded; |
| } |
| |
| @Override |
| public int getDSFID() { |
| return DLOCK_REQUEST_MESSAGE; |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| out.writeUTF(serviceName); |
| DataSerializer.writeObject(objectName, out); |
| out.writeLong(startTime); |
| out.writeLong(leaseMillis); |
| out.writeLong(waitMillis); |
| out.writeBoolean(reentrant); |
| out.writeBoolean(tryLock); |
| out.writeInt(processorId); |
| out.writeInt(lockId); |
| out.writeInt(threadId); |
| out.writeLong(grantorVersion); |
| out.writeInt(grantorSerialNumber); |
| out.writeInt(dlsSerialNumber); |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| serviceName = in.readUTF(); |
| objectName = DataSerializer.readObject(in); |
| startTime = in.readLong(); |
| leaseMillis = in.readLong(); |
| waitMillis = in.readLong(); |
| reentrant = in.readBoolean(); |
| tryLock = in.readBoolean(); |
| processorId = in.readInt(); |
| lockId = in.readInt(); |
| threadId = in.readInt(); |
| grantorVersion = in.readLong(); |
| grantorSerialNumber = in.readInt(); |
| dlsSerialNumber = in.readInt(); |
| } |
| |
| @Override |
| public String toString() { |
| |
| return "{DLockRequestMessage id=" + processorId |
| + " for " + serviceName + ":" + dlsSerialNumber |
| + " name=" + objectName |
| + " start=" + startTime |
| + " sender=" + getSender() |
| + " threadId=" + threadId |
| + " leaseMillis=" + leaseMillis |
| + " waitMillis=" + waitMillis |
| + " reentrant=" + reentrant |
| + " tryLock=" + tryLock |
| + " lockId=" + lockId |
| + " grantorVersion=" + grantorVersion |
| + " grantorSerialNumber=" + grantorSerialNumber |
| + " dlsSerialNumber=" + dlsSerialNumber |
| + "}"; |
| } |
| |
| } |
| |
| // ------------------------------------------------------------------------- |
| // DLockResponseMessage |
| // ------------------------------------------------------------------------- |
| /** |
| * This is a response to an DLockRequestMessage. A response communicates one of two things: - |
| * GRANT - you can have the lock - NOT_GRANTOR - I am not the lock grantor for this service - |
| * TIMEOUT - the lock request has timed out |
| */ |
| public static class DLockResponseMessage extends ReplyMessage { |
| |
| public static final int GRANT = 0; |
| public static final int NOT_GRANTOR = 1; |
| public static final int TIMEOUT = 2; |
| public static final int NOT_HOLDER = 3; // reentrant locking attempted |
| public static final int TRY_LOCK_FAILED = 4; // try lock failed |
| public static final int SUSPENDED = 5; // dlock has suspended locking |
| public static final int DESTROYED = 6; // requestor sent NonGrantorDestroyedMessage |
| |
| /** The name of the DistributedLockService */ |
| protected String serviceName; |
| |
| /** The object name */ |
| protected Object objectName; |
| |
| /** Specifies the results of this response */ |
| protected int responseCode = NOT_GRANTOR; // default |
| |
| /** Absolute cache time millis when the lease expires */ |
| protected long leaseExpireTime; |
| |
| /** Starts out null and then set to key that conflicted for failure */ |
| protected Object keyIfFailed; |
| |
| /** Used to match this release up with its original request */ |
| protected int lockId; |
| |
| /** The serial number of the dlock service that requested this lock */ |
| protected int dlsSerialNumber; |
| |
| // set during processing of this message... |
| /** True if the receiving node has processed this reply */ |
| protected boolean processed; |
| |
| public DLockResponseMessage() {} |
| |
| /** |
| * Need to handle race condition in which this side times out waiting for the lock before |
| * receiving a GRANT response which may already be in transit to this node. |
| */ |
| @Override |
| public void process(final DistributionManager dm, final ReplyProcessor21 processor) { |
| if (processor == null) { |
| // The processor was probably cleaned up because of memberDeparted and we need to abandon |
| return; |
| } |
| if (debugReleaseOrphanedGrant()) { |
| waitToProcessDLockResponse(dm); |
| } |
| |
| |
| // TODO - This a partial fix for bug 37158. It doesn't completely |
| // eliminate the race condition when interrupting a lock request, but |
| // it does make DistributedLockServiceTest continue to pass. |
| if (keeper.retrieve(processor.getProcessorId()) != null) { |
| super.process(dm, processor); |
| } |
| if (!processed) { |
| if (responseCode == GRANT) { |
| logger.warn("No processor found for DLockResponseMessage: {}", |
| this); |
| // got a problem... response prolly came in after client side timed out |
| releaseOrphanedGrant(dm); |
| } else { |
| logger.info("No processor found for DLockResponseMessage: {}", |
| this); |
| } |
| } |
| } |
| |
| protected boolean callReleaseProcessor(DistributionManager dm, |
| InternalDistributedMember grantor) { |
| return DLockService.callReleaseProcessor(dm, serviceName, grantor, objectName, |
| false, lockId); |
| } |
| |
| /** |
| * Releases a granted lock that was orphaned by interruption of the lock request. This also |
| * releases any lock grant for which we cannot find an active reply processor. |
| */ |
| public void releaseOrphanedGrant(DistributionManager dm) { |
| InternalDistributedMember grantor = getSender(); |
| // method is rewritten to fix bug 35252 |
| boolean released = false; |
| logger.info("Releasing orphaned grant for {}", this); |
| try { |
| while (!released) { |
| dm.getCancelCriterion().checkCancelInProgress(null); |
| try { |
| if (grantor == null) { // use grantor arg on first iteration |
| GrantorInfo gi = DLockService.checkLockGrantorInfo(serviceName, dm.getSystem()); |
| grantor = gi.getId(); |
| } |
| if (grantor == null) { // still null if elder says no one is grantor |
| // nobody knows about our zombie lock so exit |
| released = true; |
| } else { |
| released = callReleaseProcessor(dm, grantor); |
| } |
| } catch (LockGrantorDestroyedException e) { |
| // loop back around to get next lock grantor |
| } catch (IllegalStateException e) { |
| if (dm.getId().equals(grantor)) { |
| // DLockToken probably threw IllegalStateException because destroyed |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "[releaseOrphanedGrant] Local grantor threw IllegalStateException handling {}", |
| this); |
| } |
| } |
| try { |
| Thread.sleep(200); |
| } catch (InterruptedException ie) { |
| Thread.currentThread().interrupt(); |
| } |
| } finally { |
| grantor = null; |
| } |
| } |
| } finally { |
| if (released) { |
| logger.info("Handled orphaned grant with release."); |
| } else { |
| logger.info("Handled orphaned grant without release."); |
| } |
| } |
| } |
| |
| public int getLockId() { |
| return lockId; |
| } |
| |
| public int getResponseCode() { |
| return responseCode; |
| } |
| |
| public void setResponseCode(int code) { |
| responseCode = code; |
| } |
| |
| public static String responseCodeToString(int responseCode) { |
| String response = null; |
| switch (responseCode) { |
| case GRANT: |
| response = "GRANT"; |
| break; |
| case NOT_GRANTOR: |
| response = "NOT_GRANTOR"; |
| break; |
| case TIMEOUT: |
| response = "TIMEOUT"; |
| break; |
| case SUSPENDED: |
| response = "SUSPENDED"; |
| break; |
| case NOT_HOLDER: |
| response = "NOT_HOLDER"; |
| break; |
| case TRY_LOCK_FAILED: |
| response = "TRY_LOCK_FAILED"; |
| break; |
| case DESTROYED: |
| response = "DESTROYED"; |
| break; |
| default: |
| response = "UNKNOWN:" + responseCode; |
| break; |
| } |
| return response; |
| } |
| |
| @Override |
| public int getDSFID() { |
| return DLOCK_RESPONSE_MESSAGE; |
| } |
| |
| @Override |
| public void toData(DataOutput out, |
| SerializationContext context) throws IOException { |
| super.toData(out, context); |
| out.writeByte(responseCode); |
| out.writeUTF(serviceName); |
| DataSerializer.writeObject(objectName, out); |
| out.writeLong(leaseExpireTime); |
| DataSerializer.writeObject(keyIfFailed, out); |
| out.writeInt(lockId); |
| out.writeInt(dlsSerialNumber); |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| responseCode = in.readByte(); |
| serviceName = in.readUTF(); |
| objectName = DataSerializer.readObject(in); |
| leaseExpireTime = in.readLong(); |
| keyIfFailed = DataSerializer.readObject(in); |
| lockId = in.readInt(); |
| dlsSerialNumber = in.readInt(); |
| } |
| |
| @Override |
| public String toString() { |
| String response = responseCodeToString(responseCode); |
| return "DLockRequestProcessor.DLockResponseMessage " + "responding " + response |
| + "; serviceName=" + serviceName + "(version " + dlsSerialNumber + ")" + "; objectName=" |
| + objectName + "; responseCode=" + responseCode + "; keyIfFailed=" + keyIfFailed |
| + "; leaseExpireTime=" + leaseExpireTime + "; processorId=" + processorId |
| + "; lockId=" + lockId; |
| } |
| } |
| |
| @MutableForTesting |
| private static boolean debugReleaseOrphanedGrant = false; |
| private static final Object waitToProcessDLockResponseLock = new Object(); |
| @MutableForTesting |
| private static volatile boolean waitToProcessDLockResponse = false; |
| |
| public static boolean debugReleaseOrphanedGrant() { |
| return debugReleaseOrphanedGrant; |
| } |
| |
| public static void setDebugReleaseOrphanedGrant(boolean value) { |
| debugReleaseOrphanedGrant = value; |
| } |
| |
| public static void setWaitToProcessDLockResponse(boolean value) { |
| synchronized (waitToProcessDLockResponseLock) { |
| waitToProcessDLockResponse = value; |
| waitToProcessDLockResponseLock.notifyAll(); |
| } |
| } |
| |
| public static void waitToProcessDLockResponse(DistributionManager dm) { |
| synchronized (waitToProcessDLockResponseLock) { |
| while (waitToProcessDLockResponse) { |
| dm.getCancelCriterion().checkCancelInProgress(null); |
| boolean interrupted = Thread.interrupted(); |
| try { |
| logger.info("Waiting to process DLockResponseMessage"); |
| waitToProcessDLockResponseLock.wait(); |
| } catch (InterruptedException e) { |
| interrupted = true; |
| } finally { |
| if (interrupted) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| } |
| } |
| } |
| |
| } |
