| /* |
| * 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 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; |
| this.grantor = lockGrantorId.getLockGrantorMember(); |
| // this.grantorVersion = grantorVersion; |
| |
| this.request = createRequest(); |
| Assert.assertTrue(getProcessorId() > 0); |
| |
| this.request.processorId = getProcessorId(); |
| this.request.serviceName = svc.getName(); |
| this.request.objectName = objectName; |
| this.request.threadId = threadId; |
| this.request.startTime = startTime; |
| this.request.leaseMillis = leaseMillis; |
| this.request.waitMillis = waitMillis; |
| this.request.reentrant = reentrant; |
| this.request.tryLock = tryLock; |
| this.request.grantorVersion = lockGrantorId.getLockGrantorVersion(); |
| this.request.grantorSerialNumber = lockGrantorId.getLockGrantorSerialNumber(); |
| this.request.dlsSerialNumber = svc.getSerialNumber(); |
| |
| this.request.setRecipient(grantor); |
| this.disableAlerts = disableAlerts; |
| } |
| |
| protected DLockRequestMessage createRequest() { |
| return new DLockRequestMessage(); |
| } |
| |
| protected CancelCriterion getCancelCriterion(DistributionManager ignoreDM) { |
| return this.svc.getCancelCriterion(); |
| } |
| |
| boolean repliedDestroyed() { |
| if (this.response == null) { |
| return false; |
| } |
| boolean result = this.response.responseCode == DLockResponseMessage.DESTROYED; |
| return result; |
| } |
| |
| boolean repliedNotHolder() { |
| if (this.response == null) { |
| return false; |
| } |
| boolean result = this.response.responseCode == DLockResponseMessage.NOT_HOLDER; |
| return result; |
| } |
| |
| boolean repliedNotGrantor() { |
| if (this.response == null) { |
| return false; |
| } |
| boolean result = this.response.responseCode == DLockResponseMessage.NOT_GRANTOR; |
| return result; |
| } |
| |
| boolean hadNoResponse() { |
| return this.response == null; |
| } |
| |
| boolean tryLockFailed() { |
| if (this.response == null) { |
| return false; |
| } |
| boolean result = this.response.responseCode == DLockResponseMessage.TRY_LOCK_FAILED; |
| return result; |
| } |
| |
| String getResponseCodeString() { |
| if (this.response == null) |
| return null; |
| return DLockResponseMessage.responseCodeToString(this.response.responseCode); |
| } |
| |
| public DLockResponseMessage getResponse() { |
| return this.response; |
| } |
| |
| long getLeaseExpireTime() { |
| return this.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); |
| 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"); |
| } |
| this.request.setSender(this.dm.getDistributionManagerId()); |
| |
| // calls processor (this) process... |
| this.request.processLocally(this.dm); |
| } |
| |
| // remote grantor... use messaging |
| else { |
| // send the message... |
| this.dm.putOutgoing(this.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 {}", |
| (this.gotLock ? "got lock" : "failed to get lock"), this.request); |
| } |
| return this.gotLock; |
| } |
| |
| @Override |
| protected boolean allowReplyFromSender() { |
| return true; |
| } |
| |
| private boolean isLockGrantor() { |
| return this.dm.getDistributionManagerId().equals(this.grantor); |
| } |
| |
| Object getKeyIfFailed() { |
| if (this.gotLock || this.response == null) |
| return null; |
| return this.response.keyIfFailed; |
| } |
| |
| protected boolean gotLock() { |
| return this.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; |
| this.response = reply; |
| |
| if (this.response.getLockId() != this.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.", |
| this.response.getLockId()); |
| } |
| Assert.assertTrue(this.response.getLockId() == this.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); |
| } |
| this.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", |
| Integer.valueOf(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; |
| private transient volatile DistributionManager receivingDM; |
| private 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(this.receivingDM != null); |
| return getSender().equals(this.receivingDM.getId()); |
| } |
| |
| public boolean isTryLock() { |
| return this.tryLock; |
| } |
| |
| @Override |
| public int getProcessorId() { |
| return this.processorId; |
| } |
| |
| public Object getObjectName() { |
| return this.objectName; |
| } |
| |
| public long getStartTime() { |
| return this.startTime; |
| } |
| |
| public long getLeaseTime() { |
| return this.leaseMillis; |
| } |
| |
| public long getWaitTime() { |
| return this.waitMillis; |
| } |
| |
| public int getLockId() { |
| return this.lockId; |
| } |
| |
| public int getThreadId() { |
| return this.threadId; |
| } |
| |
| public long getGrantorVersion() { |
| return this.grantorVersion; |
| } |
| |
| public int getGrantorSerialNumber() { |
| return this.grantorSerialNumber; |
| } |
| |
| private final transient Object rThreadLock = new Object(); |
| |
| public RemoteThread getRemoteThread() { |
| synchronized (this.rThreadLock) { |
| if (this.rThread == null) { |
| // grantor will need RemoteThread to process this request... |
| this.rThread = new RemoteThread(getSender(), getThreadId()); |
| } |
| return this.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 { |
| this.statStart = startGrantWait(); |
| this.svc = DLockService.getInternalServiceNamed(this.serviceName); |
| if (this.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); |
| } |
| this.response = createResponse(); |
| this.response.setProcessorId(getProcessorId()); |
| this.response.setRecipient(getSender()); |
| this.response.serviceName = this.serviceName; |
| this.response.objectName = this.objectName; |
| this.response.lockId = this.lockId; |
| respondWithException(replyException); |
| } |
| } |
| } |
| |
| /** Process locally without using messaging or executor */ |
| protected void processLocally(final DistributionManager dm) { |
| this.statStart = startGrantWait(); |
| this.svc = DLockService.getInternalServiceNamed(this.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(new Runnable() { |
| @Override |
| public void run() { |
| 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 { |
| this.receivingDM = dm; |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "DLockRequestMessage.basicProcess processing <{}>", |
| this); |
| } |
| this.response = createResponse(); |
| this.response.setProcessorId(getProcessorId()); |
| this.response.setRecipient(getSender()); |
| this.response.serviceName = this.serviceName; |
| this.response.objectName = this.objectName; |
| this.response.lockId = this.lockId; |
| |
| if (waitForGrantor && this.svc != null) { |
| try { |
| this.grantor = DLockGrantor.waitForGrantor(this.svc); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| this.grantor = null; // fail it |
| } |
| } |
| |
| if (this.svc == null || this.grantor == null) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "respondWithNotGrantor this.svc={} this.grantor={}", |
| this.svc, this.grantor); |
| } |
| respondWithNotGrantor(); |
| } |
| |
| else if (this.grantor.isDestroyed()) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, "respondWithNotGrantor grantor was destroyed {}", |
| this.grantor); |
| } |
| respondWithNotGrantor(); |
| } else if (this.grantor.getVersionId() != this.grantorVersion) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "respondWithNotGrantor current version is {}; request was for {}", |
| this.grantor.getVersionId(), this.grantorVersion); |
| } |
| respondWithNotGrantor(); |
| } else if (this.svc.getSerialNumber() != this.grantorSerialNumber) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "respondWithNotGrantor current serial number is {}; request was for {}", |
| this.svc.getSerialNumber(), this.grantorSerialNumber); |
| } |
| respondWithNotGrantor(); |
| } |
| |
| // this is the grantor, so the request will be processed... |
| else { |
| this.svc.checkDestroyed(); |
| if (!this.svc.isLockGrantor()) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "respondWithNotGrantor service !isLockGrantor svc={}", this.svc); |
| } |
| respondWithNotGrantor(); |
| } |
| this.grantor.checkDestroyed(); |
| |
| // handle lock re-entry... |
| if (this.reentrant) { |
| long leaseExpireTime; |
| try { |
| leaseExpireTime = this.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 (this.grantor.isDestroyed()) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "respondWithNotGrantor grantor was destroyed grantor={}", this.grantor); |
| } |
| respondWithNotGrantor(); |
| } else { |
| try { |
| this.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={}", this.svc); |
| } |
| respondWithNotGrantor(); |
| } catch (LockServiceDestroyedException e) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "LockServiceDestroyedException respondWithNotGrantor svc={}", this.svc); |
| } |
| respondWithNotGrantor(); |
| } catch (CancelException e) { |
| if (isDebugEnabled_DLS) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "CacheClosedException respondWithNotGrantor svc={} exception = {}", this.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() { |
| this.response.responseCode = DLockResponseMessage.NOT_GRANTOR; |
| sendResponse(); |
| } |
| |
| synchronized void respondWithDestroyed() { |
| this.response.responseCode = DLockResponseMessage.DESTROYED; |
| sendResponse(); |
| } |
| |
| private synchronized void respondWithNotHolder() { |
| this.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, |
| this.grantor.displayStatus(rThread, objectName)); |
| } |
| this.response.responseCode = DLockResponseMessage.TIMEOUT; |
| sendResponse(); |
| } |
| |
| synchronized void respondWithTryLockFailed(Object keyIfFailed) { |
| this.response.responseCode = DLockResponseMessage.TRY_LOCK_FAILED; |
| this.response.keyIfFailed = keyIfFailed; |
| sendResponse(); |
| } |
| |
| synchronized void respondWithGrant(long leaseExpireTime) { |
| this.response.responseCode = DLockResponseMessage.GRANT; |
| this.response.leaseExpireTime = leaseExpireTime; |
| this.response.dlsSerialNumber = this.dlsSerialNumber; |
| sendResponse(); |
| } |
| |
| synchronized void respondWithException(Throwable t) { |
| try { |
| if (this.response.getException() == null) { |
| this.response.setException(new ReplyException(t)); |
| if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "While processing <{}>, got exception, returning to sender", this, |
| 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 (this.waitMillis == -1 || this.waitMillis == Long.MAX_VALUE) { |
| return Long.MAX_VALUE; |
| } else { |
| long result = this.startTime + this.waitMillis; |
| if (result < this.startTime) { |
| result = Long.MAX_VALUE; |
| } |
| return result; |
| } |
| } |
| |
| synchronized boolean checkForTimeout() { |
| if (this.waitMillis == -1 || this.waitMillis == Long.MAX_VALUE) |
| return false; |
| if (this.tryLock) |
| return false; |
| long now = DLockService.getLockTimeStamp(this.receivingDM); |
| if (now < this.startTime) |
| now = this.startTime; |
| if (this.waitMillis + this.startTime - now <= 0) { |
| if (logger.isTraceEnabled(LogMarker.DLS_VERBOSE)) { |
| logger.trace(LogMarker.DLS_VERBOSE, |
| "DLockRequestProcessor request timed out: waitMillis={} now={} startTime={}", |
| this.waitMillis, now, this.startTime); |
| } |
| respondWithTimeout(); |
| return true; |
| } |
| return false; |
| } |
| |
| private void endGrantWaitStatistic() { |
| if (this.statStart == -1) |
| return; // failed to start the stat |
| DistributedLockStats stats = DLockService.getDistributedLockStats(); |
| switch (this.response.responseCode) { |
| case DLockResponseMessage.GRANT: |
| stats.endGrantWait(this.statStart); |
| break; |
| case DLockResponseMessage.NOT_GRANTOR: |
| stats.endGrantWaitNotGrantor(this.statStart); |
| break; |
| case DLockResponseMessage.TIMEOUT: |
| stats.endGrantWaitTimeout(this.statStart); |
| break; |
| case DLockResponseMessage.SUSPENDED: |
| stats.endGrantWaitSuspended(this.statStart); |
| break; |
| case DLockResponseMessage.NOT_HOLDER: |
| stats.endGrantWaitNotHolder(this.statStart); |
| break; |
| case DLockResponseMessage.TRY_LOCK_FAILED: |
| stats.endGrantWaitFailed(this.statStart); |
| break; |
| case DLockResponseMessage.DESTROYED: |
| stats.endGrantWaitDestroyed(this.statStart); |
| break; |
| default: |
| Assert.assertTrue(false, "Unknown responseCode: " + this.response.responseCode); |
| break; |
| } |
| } |
| |
| /** Callers must be synchronized on this */ |
| private void sendResponse() { |
| try { |
| if (this.responded) |
| return; |
| InternalDistributedMember myId = this.receivingDM.getDistributionManagerId(); |
| |
| // local... don't actually use messaging |
| if (getSender().equals(myId)) { |
| if (debugReleaseOrphanedGrant()) { |
| waitToProcessDLockResponse(this.receivingDM); |
| } |
| ReplyProcessor21 processor = ReplyProcessor21.getProcessor(processorId); |
| if (processor == null) { |
| // lock request was probably interrupted so we need to release it... |
| logger.warn(LogMarker.DLS_MARKER, |
| "Failed to find processor for {}", |
| this.response); |
| if (this.response.responseCode == DLockResponseMessage.GRANT) { |
| logger.info(LogMarker.DLS_MARKER, |
| "Releasing local orphaned grant for {}.", |
| this); |
| try { |
| this.grantor.releaseIfLocked(this.objectName, getSender(), this.lockId); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| logger.warn("Interrupted while releasing grant.", e); |
| } |
| logger.info("Handled local orphaned grant."); |
| } |
| endGrantWaitStatistic(); |
| return; |
| } |
| this.response.setSender(getSender()); |
| endGrantWaitStatistic(); |
| processor.process(this.response); |
| } |
| |
| // remote... use messaging |
| else { |
| this.receivingDM.putOutgoing(this.response); |
| endGrantWaitStatistic(); |
| } |
| } finally { |
| this.responded = true; |
| } |
| } |
| |
| synchronized void handleDepartureOfSender() { |
| try { |
| if (this.receivingDM.getDistributionManagerIds().contains(this.sender)) { |
| // sender must have sent us a NonGrantorDestroyedMessage |
| // still need to send a reply to make the thread stop waiting |
| respondWithDestroyed(); |
| } |
| } finally { |
| if (!this.responded) { |
| endGrantWaitStatistic(); |
| this.responded = true; |
| } |
| } |
| } |
| |
| synchronized boolean responded() { |
| return this.responded; |
| } |
| |
| /** Callers must be synchronized on this */ |
| boolean respondedNoSync() { |
| return this.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(this.serviceName); |
| DataSerializer.writeObject(this.objectName, out); |
| out.writeLong(this.startTime); |
| out.writeLong(this.leaseMillis); |
| out.writeLong(this.waitMillis); |
| out.writeBoolean(this.reentrant); |
| out.writeBoolean(this.tryLock); |
| out.writeInt(this.processorId); |
| out.writeInt(this.lockId); |
| out.writeInt(this.threadId); |
| out.writeLong(this.grantorVersion); |
| out.writeInt(this.grantorSerialNumber); |
| out.writeInt(this.dlsSerialNumber); |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| this.serviceName = in.readUTF(); |
| this.objectName = DataSerializer.readObject(in); |
| this.startTime = in.readLong(); |
| this.leaseMillis = in.readLong(); |
| this.waitMillis = in.readLong(); |
| this.reentrant = in.readBoolean(); |
| this.tryLock = in.readBoolean(); |
| this.processorId = in.readInt(); |
| this.lockId = in.readInt(); |
| this.threadId = in.readInt(); |
| this.grantorVersion = in.readLong(); |
| this.grantorSerialNumber = in.readInt(); |
| this.dlsSerialNumber = in.readInt(); |
| } |
| |
| @Override |
| public String toString() { |
| StringBuffer sb = new StringBuffer(); |
| |
| sb.append("{DLockRequestMessage id=" + this.processorId); |
| sb.append(" for " + this.serviceName + ":" + this.dlsSerialNumber); |
| sb.append(" name=" + this.objectName); |
| sb.append(" start=" + this.startTime); |
| sb.append(" sender=" + getSender()); |
| sb.append(" threadId=" + this.threadId); |
| sb.append(" leaseMillis=" + this.leaseMillis); |
| sb.append(" waitMillis=" + this.waitMillis); |
| sb.append(" reentrant=" + this.reentrant); |
| sb.append(" tryLock=" + this.tryLock); |
| sb.append(" lockId=" + this.lockId); |
| sb.append(" grantorVersion=" + this.grantorVersion); |
| sb.append(" grantorSerialNumber=" + this.grantorSerialNumber); |
| sb.append(" dlsSerialNumber=" + this.dlsSerialNumber); |
| sb.append("}"); |
| return sb.toString(); |
| } |
| |
| } |
| |
| // ------------------------------------------------------------------------- |
| // 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 (!this.processed) { |
| if (this.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, this.serviceName, grantor, this.objectName, |
| false, this.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(this.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 this.lockId; |
| } |
| |
| public int getResponseCode() { |
| return this.responseCode; |
| } |
| |
| public void setResponseCode(int code) { |
| this.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:" + String.valueOf(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(this.responseCode); |
| out.writeUTF(this.serviceName); |
| DataSerializer.writeObject(this.objectName, out); |
| out.writeLong(this.leaseExpireTime); |
| DataSerializer.writeObject(this.keyIfFailed, out); |
| out.writeInt(this.lockId); |
| out.writeInt(this.dlsSerialNumber); |
| } |
| |
| @Override |
| public void fromData(DataInput in, |
| DeserializationContext context) throws IOException, ClassNotFoundException { |
| super.fromData(in, context); |
| this.responseCode = in.readByte(); |
| this.serviceName = in.readUTF(); |
| this.objectName = DataSerializer.readObject(in); |
| this.leaseExpireTime = in.readLong(); |
| this.keyIfFailed = DataSerializer.readObject(in); |
| this.lockId = in.readInt(); |
| this.dlsSerialNumber = in.readInt(); |
| } |
| |
| @Override |
| public String toString() { |
| String response = responseCodeToString(this.responseCode); |
| return "DLockRequestProcessor.DLockResponseMessage " + "responding " + response |
| + "; serviceName=" + serviceName + "(version " + dlsSerialNumber + ")" + "; objectName=" |
| + objectName + "; responseCode=" + responseCode + "; keyIfFailed=" + keyIfFailed |
| + "; leaseExpireTime=" + leaseExpireTime + "; processorId=" + this.processorId |
| + "; lockId=" + this.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(); |
| } |
| } |
| } |
| } |
| |
| } |
