| /* |
| * 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.internal.cache.tier.sockets; |
| |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.InterruptedIOException; |
| import java.io.OutputStream; |
| import java.net.ConnectException; |
| import java.net.Socket; |
| import java.net.SocketException; |
| import java.nio.ByteBuffer; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| |
| import javax.net.ssl.SSLException; |
| |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.InvalidDeltaException; |
| import org.apache.geode.StatisticDescriptor; |
| import org.apache.geode.Statistics; |
| import org.apache.geode.StatisticsType; |
| import org.apache.geode.StatisticsTypeFactory; |
| import org.apache.geode.annotations.Immutable; |
| import org.apache.geode.annotations.internal.MutableForTesting; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.InterestResultPolicy; |
| import org.apache.geode.cache.Operation; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.client.ServerRefusedConnectionException; |
| import org.apache.geode.cache.client.internal.ClientUpdater; |
| import org.apache.geode.cache.client.internal.Endpoint; |
| import org.apache.geode.cache.client.internal.EndpointManager; |
| import org.apache.geode.cache.client.internal.GetEventValueOp; |
| import org.apache.geode.cache.client.internal.PoolImpl; |
| import org.apache.geode.cache.client.internal.QueueManager; |
| import org.apache.geode.cache.query.internal.cq.CqService; |
| import org.apache.geode.distributed.DistributedSystem; |
| import org.apache.geode.distributed.internal.DistributionConfig; |
| import org.apache.geode.distributed.internal.DistributionStats; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem.DisconnectListener; |
| import org.apache.geode.distributed.internal.ServerLocation; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.Assert; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.InternalInstantiator; |
| import org.apache.geode.internal.cache.ClientServerObserver; |
| import org.apache.geode.internal.cache.ClientServerObserverHolder; |
| import org.apache.geode.internal.cache.EntryEventImpl; |
| import org.apache.geode.internal.cache.EventID; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.tier.CachedRegionHelper; |
| import org.apache.geode.internal.cache.tier.ClientSideHandshake; |
| import org.apache.geode.internal.cache.tier.MessageType; |
| import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException; |
| import org.apache.geode.internal.cache.versions.VersionSource; |
| import org.apache.geode.internal.cache.versions.VersionTag; |
| import org.apache.geode.internal.logging.InternalLogWriter; |
| import org.apache.geode.internal.logging.log4j.LogMarker; |
| import org.apache.geode.internal.net.SocketCreator; |
| import org.apache.geode.internal.offheap.annotations.Released; |
| import org.apache.geode.internal.sequencelog.EntryLogger; |
| import org.apache.geode.internal.serialization.ByteArrayDataInput; |
| import org.apache.geode.internal.serialization.Version; |
| import org.apache.geode.internal.statistics.StatisticsTypeFactoryImpl; |
| import org.apache.geode.logging.internal.executors.LoggingThread; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| import org.apache.geode.security.AuthenticationFailedException; |
| import org.apache.geode.security.AuthenticationRequiredException; |
| import org.apache.geode.security.GemFireSecurityException; |
| |
| /** |
| * {@code CacheClientUpdater} is a thread that processes update messages from a cache server and |
| * {@linkplain org.apache.geode.cache.Region#localInvalidate(Object) invalidates} the local cache |
| * based on the contents of those messages. |
| * |
| * @since GemFire 3.5 |
| */ |
| public class CacheClientUpdater extends LoggingThread implements ClientUpdater, DisconnectListener { |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| private static final int DEFAULT_SOCKET_BUFFER_SIZE = 32768; |
| |
| /** |
| * true if the constructor successfully created a connection. If false, the run method for this |
| * thread immediately exits. |
| */ |
| private final boolean connected; |
| |
| /** |
| * System of which we are a part |
| */ |
| private final DistributedSystem system; |
| |
| /** |
| * The socket by which we communicate with the server |
| */ |
| private final Socket socket; |
| |
| /** |
| * The output stream of the socket |
| */ |
| private final OutputStream out; |
| |
| /** |
| * The input stream of the socket |
| */ |
| private final InputStream in; |
| |
| public ServerQueueStatus getServerQueueStatus() { |
| return serverQueueStatus; |
| } |
| |
| /** |
| * server-side queue status at the time we connected to it |
| */ |
| private ServerQueueStatus serverQueueStatus; |
| |
| /** |
| * Failed updater from the endpoint previously known as the primary |
| */ |
| private volatile ClientUpdater failedUpdater; |
| |
| /** |
| * The buffer upon which we receive messages |
| */ |
| private final ByteBuffer commBuffer; |
| |
| private boolean commBufferReleased; // TODO: fix synchronization |
| |
| private final CCUStats stats; |
| |
| /** |
| * Cache for which we provide service TODO: lifecycle and synchronization need work |
| */ |
| private /* final */ InternalCache cache; |
| |
| private /* final */ CachedRegionHelper cacheHelper; |
| |
| /** |
| * Principle flag to signal thread's run loop to terminate |
| */ |
| private final AtomicBoolean continueProcessing = new AtomicBoolean(true); |
| |
| /** |
| * Is the client durable Used for bug 39010 fix |
| */ |
| private final boolean isDurableClient; |
| |
| /** |
| * Represents the server we are connected to |
| */ |
| private final InternalDistributedMember serverId; |
| |
| /** |
| * true if the EndPoint represented by this updater thread is primary |
| */ |
| private final boolean isPrimary; |
| |
| /** |
| * Added to avoid recording of the event if the concerned operation failed. See #43247 |
| */ |
| private boolean isOpCompleted; |
| |
| public static final String CLIENT_UPDATER_THREAD_NAME = "Cache Client Updater Thread "; |
| |
| /** |
| * to enable test flag TODO: eliminate isUsedByTest |
| */ |
| @MutableForTesting |
| public static boolean isUsedByTest; |
| |
| /** |
| * Indicates if full value was requested from server as a result of failure in applying delta |
| * bytes. TODO: only used for test assertion |
| */ |
| @MutableForTesting |
| static boolean fullValueRequested = false; |
| |
| private final ServerLocation location; |
| |
| // TODO - remove these fields |
| private QueueManager qManager = null; |
| private EndpointManager eManager = null; |
| private Endpoint endpoint = null; |
| |
| private static final long MAX_CACHE_WAIT = |
| Long.getLong(DistributionConfig.GEMFIRE_PREFIX + "CacheClientUpdater.MAX_WAIT", 120); // seconds |
| |
| /** |
| * Return true if cache appears |
| * |
| * @return true if cache appears |
| */ |
| private boolean waitForCache() { |
| InternalCache cache; |
| long tilt = System.currentTimeMillis() + MAX_CACHE_WAIT * 1000; |
| for (;;) { |
| if (quitting()) { |
| logger.warn("{}: abandoned wait due to cancellation.", this); |
| return false; |
| } |
| if (!this.connected) { |
| logger.warn("{}: abandoned wait because it is no longer connected", |
| this); |
| return false; |
| } |
| if (System.currentTimeMillis() > tilt) { |
| logger.warn("{}: wait timed out (more than {} seconds)", |
| new Object[] {this, MAX_CACHE_WAIT}); |
| return false; |
| } |
| cache = GemFireCacheImpl.getInstance(); |
| if (cache != null && !cache.isClosed()) { |
| break; |
| } |
| boolean interrupted = Thread.interrupted(); |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException ignore) { |
| interrupted = true; |
| } finally { |
| if (interrupted) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| } // for |
| this.cache = cache; |
| this.cacheHelper = new CachedRegionHelper(cache); |
| return true; |
| } |
| |
| /** |
| * Creates a new {@code CacheClientUpdater} with a given name that waits for a server to connect |
| * on a given port. |
| * |
| * @param name descriptive name, used for our ThreadGroup |
| * @param location the endpoint we represent |
| * @param primary true if our endpoint is primary |
| * @param ids the system we are distributing messages through |
| * |
| * @throws AuthenticationRequiredException when client is not configured to send credentials using |
| * security-* system properties but server expects credentials |
| * @throws AuthenticationFailedException when authentication of the client fails |
| * @throws ServerRefusedConnectionException when handshake fails for other reasons like using |
| * durable client ID that is already in use by another client or some server side |
| * exception while reading handshake/verifying credentials |
| */ |
| public CacheClientUpdater(String name, ServerLocation location, boolean primary, |
| DistributedSystem ids, ClientSideHandshake handshake, QueueManager qManager, |
| EndpointManager eManager, Endpoint endpoint, int handshakeTimeout, |
| SocketCreator socketCreator) throws AuthenticationRequiredException, |
| AuthenticationFailedException, ServerRefusedConnectionException { |
| this(name, location, primary, ids, handshake, qManager, eManager, endpoint, handshakeTimeout, |
| socketCreator, new StatisticsProvider()); |
| } |
| |
| /** |
| * alternative constructor for unit tests. This constructor allows you to pass a |
| * mock StatisticsProvider |
| */ |
| public CacheClientUpdater(String name, ServerLocation location, boolean primary, |
| DistributedSystem distributedSystem, ClientSideHandshake handshake, QueueManager qManager, |
| EndpointManager eManager, Endpoint endpoint, int handshakeTimeout, |
| SocketCreator socketCreator, StatisticsProvider statisticsProvider) |
| throws AuthenticationRequiredException, |
| AuthenticationFailedException, ServerRefusedConnectionException { |
| super(name); |
| this.system = distributedSystem; |
| this.isDurableClient = handshake.isDurable(); |
| this.isPrimary = primary; |
| this.location = location; |
| this.qManager = qManager; |
| // this holds the connection which this threads reads |
| this.eManager = eManager; |
| this.endpoint = endpoint; |
| this.stats = statisticsProvider.createStatistics(distributedSystem, location); |
| |
| // Create the connection... |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| if (isDebugEnabled) { |
| logger.debug("Creating asynchronous update connection"); |
| } |
| |
| boolean success = false; |
| Socket mySock = null; |
| InternalDistributedMember sid = null; |
| ByteBuffer cb = null; |
| OutputStream tmpOut = null; |
| InputStream tmpIn = null; |
| try { |
| // Size of the server-to-client communication socket buffers |
| int socketBufferSize = |
| Integer.getInteger("BridgeServer.SOCKET_BUFFER_SIZE", DEFAULT_SOCKET_BUFFER_SIZE); |
| |
| mySock = socketCreator.connectForClient(location.getHostName(), location.getPort(), |
| handshakeTimeout, socketBufferSize); |
| mySock.setTcpNoDelay(true); |
| mySock.setSendBufferSize(socketBufferSize); |
| |
| // Verify buffer sizes |
| verifySocketBufferSize(socketBufferSize, mySock.getReceiveBufferSize(), "receive"); |
| verifySocketBufferSize(socketBufferSize, mySock.getSendBufferSize(), "send"); |
| |
| // set the timeout for the handshake |
| mySock.setSoTimeout(handshakeTimeout); |
| tmpOut = mySock.getOutputStream(); |
| tmpIn = mySock.getInputStream(); |
| |
| if (isDebugEnabled) { |
| logger.debug( |
| "Initialized server-to-client socket with send buffer size: {} bytes and receive buffer size: {} bytes", |
| mySock.getSendBufferSize(), mySock.getReceiveBufferSize()); |
| } |
| |
| if (isDebugEnabled) { |
| logger.debug( |
| "Created connection from {}:{} to CacheClientNotifier on port {} for server-to-client communication", |
| mySock.getInetAddress().getHostAddress(), mySock.getLocalPort(), mySock.getPort()); |
| } |
| |
| this.serverQueueStatus = handshake.handshakeWithSubscriptionFeed(mySock, this.isPrimary); |
| if (serverQueueStatus.isPrimary() || serverQueueStatus.isNonRedundant()) { |
| PoolImpl pool = (PoolImpl) this.qManager.getPool(); |
| if (!pool.getReadyForEventsCalled()) { |
| pool.setPendingEventCount(serverQueueStatus.getServerQueueSize()); |
| } |
| } |
| |
| int bufSize = 1024; |
| try { |
| bufSize = mySock.getSendBufferSize(); |
| if (bufSize < 1024) { |
| bufSize = 1024; |
| } |
| } catch (SocketException ignore) { |
| } |
| cb = ServerConnection.allocateCommBuffer(bufSize, mySock); |
| |
| sid = |
| new InternalDistributedMember(mySock.getInetAddress(), mySock.getPort(), false, true); |
| |
| success = true; |
| } catch (ConnectException ignore) { |
| if (!quitting()) { |
| logger.warn("{} connection was refused", this); |
| } |
| } catch (SSLException ex) { |
| if (!quitting()) { |
| getSecurityLogger().warning(String.format("%s SSL negotiation failed. %s", |
| new Object[] {this, ex})); |
| throw new AuthenticationFailedException( |
| String.format("SSL negotiation failed with endpoint: %s", |
| location), |
| ex); |
| } |
| } catch (GemFireSecurityException ex) { |
| if (!quitting()) { |
| getSecurityLogger().warning( |
| String.format( |
| "%s: Security exception when creating server-to-client communication socket. %s", |
| new Object[] {this, ex})); |
| throw ex; |
| } |
| } catch (IOException e) { |
| if (!quitting()) { |
| logger.warn(String.format( |
| "%s: Caught following exception while attempting to create a server-to-client communication socket and will exit: %s", |
| new Object[] {this, e}), |
| logger.isDebugEnabled() ? e : null); |
| } |
| eManager.serverCrashed(this.endpoint); |
| } catch (ClassNotFoundException e) { |
| if (!quitting()) { |
| logger.warn("Unable to load the class: {}", |
| e.getMessage()); |
| } |
| } catch (ServerRefusedConnectionException e) { |
| if (!quitting()) { |
| logger.warn(String.format( |
| "%s: Caught following exception while attempting to create a server-to-client communication socket and will exit: %s", |
| new Object[] {this, e}), |
| logger.isDebugEnabled() ? e : null); |
| } |
| throw e; |
| } finally { |
| this.connected = success; |
| this.socket = mySock; |
| this.commBuffer = cb; |
| this.out = tmpOut; |
| this.in = tmpIn; |
| this.serverId = sid; |
| if (this.connected) { |
| if (mySock != null) { |
| try { |
| mySock.setSoTimeout(0); |
| } catch (SocketException ignore) { |
| // ignore: nothing we can do about this |
| } |
| } |
| } else { |
| close(); |
| } |
| } |
| } |
| |
| private void releaseCommBuffer() { |
| if (!this.commBufferReleased) { |
| if (this.commBuffer != null) { |
| synchronized (this.commBuffer) { |
| if (!this.commBufferReleased) { |
| this.commBufferReleased = true; |
| ServerConnection.releaseCommBuffer(this.commBuffer); |
| } |
| } |
| } |
| } |
| } |
| |
| public boolean isConnected() { |
| return this.connected; |
| } |
| |
| @Override |
| public boolean isPrimary() { |
| return this.isPrimary; |
| } |
| |
| public InternalLogWriter getSecurityLogger() { |
| return this.qManager.getSecurityLogger(); |
| } |
| |
| @Override |
| public void setFailedUpdater(ClientUpdater failedUpdater) { |
| this.failedUpdater = failedUpdater; |
| } |
| |
| /** |
| * Performs the work of the client update thread. Creates a {@code ServerSocket} and waits for the |
| * server to connect to it. |
| */ |
| @Override |
| public void run() { |
| EntryLogger.setSource(this.serverId, "RI"); |
| boolean addedListener = false; |
| try { |
| if (system instanceof InternalDistributedSystem) { |
| ((InternalDistributedSystem) system).addDisconnectListener(this); |
| addedListener = true; |
| } |
| |
| if (!waitForCache()) { |
| logger.warn("{}: no cache (exiting)", this); |
| return; |
| } |
| processMessages(); |
| |
| } catch (CancelException ignore) { |
| // just bail |
| |
| } finally { |
| if (addedListener) { |
| ((InternalDistributedSystem) system).removeDisconnectListener(this); |
| } |
| this.close(); |
| EntryLogger.clearSource(); |
| } |
| } |
| |
| /** |
| * Notifies this thread to stop processing |
| */ |
| private void stopProcessing() { |
| this.continueProcessing.set(false); |
| } |
| |
| /** |
| * Stops the updater. It will wait for a while for the thread to finish to try to prevent |
| * duplicates. Note: this method is not named stop because this is a Thread which has a deprecated |
| * stop method. |
| */ |
| private void stopUpdater() { |
| boolean isSelfDestroying = Thread.currentThread() == this; |
| stopProcessing(); |
| |
| // need to also close the socket for this interrupt to wakeup |
| // the thread. This fixes bug 35691. |
| |
| if (this.isAlive()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Stopping {}", this.location, this); |
| } |
| |
| if (!isSelfDestroying) { |
| interrupt(); |
| try { |
| if (this.socket != null) { |
| this.socket.close(); |
| } |
| } catch (IOException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug(e.getMessage(), e); |
| } |
| } |
| } // !isSelfDestroying |
| } // isAlive |
| } |
| |
| /** |
| * Signals the run thread to stop, closes underlying resources. |
| */ |
| @Override |
| public void close() { |
| this.continueProcessing.set(false); // signals we are done. |
| |
| // Close the socket. This will also cause the underlying streams to fail. |
| try { |
| if (this.socket != null) { |
| this.socket.close(); |
| } |
| } catch (IOException ignore) { |
| // ignore |
| } |
| |
| if (this.cacheHelper != null) { |
| this.cacheHelper.close(); |
| } |
| releaseCommBuffer(); |
| |
| this.stats.close(); |
| } |
| |
| /** |
| * Creates a cached {@link Message}object whose state is filled in with a message received from |
| * the server. |
| */ |
| private Message initializeMessage() { |
| Message message = new Message(2, Version.CURRENT); |
| message.setComms(this.socket, this.in, this.out, this.commBuffer, this.stats); |
| return message; |
| } |
| |
| /* refinement of method inherited from Thread */ |
| @Override |
| public String toString() { |
| return getName() + " (" + this.location.getHostName() + ':' + this.location.getPort() + ')'; |
| } |
| |
| /** |
| * Handle a marker message |
| * |
| * @param clientMessage message containing the data |
| */ |
| private void handleMarker(Message clientMessage) { |
| try { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| if (isDebugEnabled) { |
| logger.debug("Received marker message of length ({} bytes)", |
| clientMessage.getPayloadLength()); |
| } |
| |
| this.qManager.getState().processMarker(); |
| |
| if (isDebugEnabled) { |
| logger.debug("Processed marker message"); |
| } |
| } catch (Exception e) { |
| String message = |
| "The following exception occurred while attempting to handle a marker."; |
| handleException(message, e); |
| } |
| } |
| |
| /** |
| * Create or update an entry |
| * |
| * @param clientMessage message containing the data |
| */ |
| private void handleUpdate(Message clientMessage) { |
| String regionName = null; |
| Object key = null; |
| Part valuePart = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| this.isOpCompleted = false; |
| |
| // Retrieve the data from the put message parts |
| if (isDebugEnabled) { |
| logger.debug("Received put message of length ({} bytes)", clientMessage.getPayloadLength()); |
| } |
| |
| int partCnt = 0; |
| Part regionNamePart = clientMessage.getPart(partCnt++); |
| Part keyPart = clientMessage.getPart(partCnt++); |
| boolean isDeltaSent = (Boolean) clientMessage.getPart(partCnt++).getObject(); |
| valuePart = clientMessage.getPart(partCnt++); |
| Part callbackArgumentPart = clientMessage.getPart(partCnt++); |
| VersionTag versionTag = (VersionTag) clientMessage.getPart(partCnt++).getObject(); |
| if (versionTag != null) { |
| versionTag.replaceNullIDs((InternalDistributedMember) this.endpoint.getMemberId()); |
| } |
| Part isInterestListPassedPart = clientMessage.getPart(partCnt++); |
| Part hasCqsPart = clientMessage.getPart(partCnt++); |
| |
| EventID eventId = |
| (EventID) clientMessage.getPart(clientMessage.getNumberOfParts() - 1).getObject(); |
| |
| boolean withInterest = (Boolean) isInterestListPassedPart.getObject(); |
| boolean withCQs = (Boolean) hasCqsPart.getObject(); |
| |
| regionName = regionNamePart.getCachedString(); |
| key = keyPart.getStringOrObject(); |
| Object callbackArgument = callbackArgumentPart.getObject(); |
| |
| // Don't automatically deserialize the value. |
| // Pass it onto the region as a byte[]. If it is a serialized |
| // object, it will be stored as a CachedDeserializable and |
| // deserialized only when requested. |
| |
| boolean isCreate = clientMessage.getMessageType() == MessageType.LOCAL_CREATE; |
| |
| if (isDebugEnabled) { |
| logger.debug( |
| "Putting entry for region: {} key: {} create: {}{} callbackArgument: {} withInterest={} withCQs={} eventID={} version={}", |
| regionName, key, isCreate, |
| valuePart.isObject() |
| ? new StringBuilder(" value: ").append(deserialize(valuePart.getSerializedForm())) |
| : "", |
| callbackArgument, withInterest, withCQs, eventId, versionTag); |
| } |
| |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| |
| Object newValue = null; |
| byte[] deltaBytes = null; |
| Object fullValue = null; |
| boolean isValueObject; |
| |
| if (!isDeltaSent) { |
| // bug #42162 - must check for a serialized null here |
| byte[] serializedForm = valuePart.getSerializedForm(); |
| |
| if (isCreate && InternalDataSerializer.isSerializedNull(serializedForm)) { |
| // newValue = null; newValue is already null |
| } else { |
| newValue = valuePart.getSerializedForm(); |
| } |
| |
| if (withCQs) { |
| fullValue = valuePart.getObject(); |
| } |
| |
| isValueObject = valuePart.isObject(); |
| } else { |
| deltaBytes = valuePart.getSerializedForm(); |
| isValueObject = true; |
| } |
| |
| if (region == null) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug("{}: Region named {} does not exist", this, regionName); |
| } |
| |
| } else if (region.hasServerProxy() && ServerResponseMatrix |
| .checkForValidStateAfterNotification(region, key, clientMessage.getMessageType()) |
| && (withInterest || !withCQs)) { |
| @Released |
| EntryEventImpl newEvent = null; |
| |
| try { |
| // Create an event and put the entry |
| newEvent = EntryEventImpl.create(region, |
| clientMessage.getMessageType() == MessageType.LOCAL_CREATE ? Operation.CREATE |
| : Operation.UPDATE, |
| key, null /* newValue */, callbackArgument /* callbackArg */, true /* originRemote */, |
| eventId.getDistributedMember()); |
| |
| newEvent.setVersionTag(versionTag); |
| newEvent.setFromServer(true); |
| |
| region.basicBridgeClientUpdate(eventId.getDistributedMember(), key, newValue, deltaBytes, |
| isValueObject, callbackArgument, |
| clientMessage.getMessageType() == MessageType.LOCAL_CREATE, |
| this.qManager.getState().getProcessedMarker() || !this.isDurableClient, newEvent, |
| eventId); |
| |
| this.isOpCompleted = true; |
| |
| // bug 45520 - ConcurrentCacheModificationException is not thrown and we must check this |
| // flag |
| if (withCQs && isDeltaSent) { |
| fullValue = newEvent.getNewValue(); |
| } |
| } catch (InvalidDeltaException ignore) { |
| Part fullValuePart = requestFullValue(eventId, "Caught InvalidDeltaException."); |
| region.getCachePerfStats().incDeltaFullValuesRequested(); |
| fullValue = newValue = fullValuePart.getObject(); // TODO: fix this line |
| isValueObject = fullValuePart.isObject(); |
| |
| region.basicBridgeClientUpdate(eventId.getDistributedMember(), key, newValue, null, |
| isValueObject, callbackArgument, |
| clientMessage.getMessageType() == MessageType.LOCAL_CREATE, |
| this.qManager.getState().getProcessedMarker() || !this.isDurableClient, newEvent, |
| eventId); |
| |
| this.isOpCompleted = true; |
| } finally { |
| if (newEvent != null) |
| newEvent.release(); |
| } |
| |
| if (isDebugEnabled) { |
| logger.debug("Put entry for region: {} key: {} callbackArgument: {}", regionName, key, |
| callbackArgument); |
| } |
| } |
| |
| // Update CQs. CQs can exist without client region. |
| if (withCQs) { |
| Part numCqsPart = clientMessage.getPart(partCnt++); |
| if (isDebugEnabled) { |
| logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}", |
| numCqsPart.getInt() / 2); |
| } |
| partCnt = processCqs(clientMessage, partCnt, numCqsPart.getInt(), |
| clientMessage.getMessageType(), key, fullValue, deltaBytes, eventId); |
| this.isOpCompleted = true; |
| } |
| } catch (Exception e) { |
| String message = |
| String.format( |
| "The following exception occurred while attempting to put entry (region: %s key: %s value: %s)", |
| regionName, key, deserialize(valuePart.getSerializedForm())); |
| handleException(message, e); |
| } |
| } |
| |
| private Part requestFullValue(EventID eventId, String reason) throws Exception { |
| if (isUsedByTest) { |
| fullValueRequested = true; |
| } |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| if (isDebugEnabled) { |
| logger.debug("{} Requesting full value...", reason); |
| } |
| Part result = (Part) GetEventValueOp.executeOnPrimary(this.qManager.getPool(), eventId, null); |
| |
| if (result == null) { |
| // Just log a warning. Do not stop CCU thread. |
| // TODO: throw a subclass of Exception |
| throw new Exception("Could not retrieve full value for " + eventId); |
| } |
| |
| if (isDebugEnabled) { |
| logger.debug("Full value received."); |
| } |
| return result; |
| } |
| |
| /** |
| * Invalidate an entry |
| * |
| * @param clientMessage message describing the entry |
| */ |
| private void handleInvalidate(Message clientMessage) { |
| String regionName = null; |
| Object key = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| this.isOpCompleted = false; |
| |
| // Retrieve the data from the local-invalidate message parts |
| if (isDebugEnabled) { |
| logger.debug("Received invalidate message of length ({} bytes)", |
| clientMessage.getPayloadLength()); |
| } |
| |
| int partCnt = 0; |
| Part regionNamePart = clientMessage.getPart(partCnt++); |
| Part keyPart = clientMessage.getPart(partCnt++); |
| Part callbackArgumentPart = clientMessage.getPart(partCnt++); |
| |
| VersionTag versionTag = (VersionTag) clientMessage.getPart(partCnt++).getObject(); |
| if (versionTag != null) { |
| versionTag.replaceNullIDs((InternalDistributedMember) this.endpoint.getMemberId()); |
| } |
| |
| Part isInterestListPassedPart = clientMessage.getPart(partCnt++); |
| Part hasCqsPart = clientMessage.getPart(partCnt++); |
| |
| regionName = regionNamePart.getCachedString(); |
| key = keyPart.getStringOrObject(); |
| |
| Object callbackArgument = callbackArgumentPart.getObject(); |
| boolean withInterest = (Boolean) isInterestListPassedPart.getObject(); |
| boolean withCQs = (Boolean) hasCqsPart.getObject(); |
| |
| if (isDebugEnabled) { |
| logger.debug( |
| "Invalidating entry for region: {} key: {} callbackArgument: {} withInterest={} withCQs={} version={}", |
| regionName, key, callbackArgument, withInterest, withCQs, versionTag); |
| } |
| |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| if (region == null) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug("Region named {} does not exist", regionName); |
| } |
| |
| } else { |
| if (region.hasServerProxy() && (withInterest || !withCQs)) { |
| try { |
| Part eid = clientMessage.getPart(clientMessage.getNumberOfParts() - 1); |
| EventID eventId = (EventID) eid.getObject(); |
| |
| try { |
| region.basicBridgeClientInvalidate(eventId.getDistributedMember(), key, |
| callbackArgument, |
| this.qManager.getState().getProcessedMarker() || !this.isDurableClient, eventId, |
| versionTag); |
| } catch (ConcurrentCacheModificationException ignore) { |
| // allow CQs to be processed |
| } |
| |
| this.isOpCompleted = true; |
| // fix for 36615 |
| this.qManager.getState().incrementInvalidatedStats(); |
| |
| if (isDebugEnabled) { |
| logger.debug("Invalidated entry for region: {} key: {} callbackArgument: {}", |
| regionName, key, callbackArgument); |
| } |
| } catch (EntryNotFoundException ignore) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug("Already invalidated entry for region: {} key: {} callbackArgument: {}", |
| regionName, key, callbackArgument); |
| } |
| this.isOpCompleted = true; |
| } |
| } |
| } |
| |
| if (withCQs) { |
| // The client may have been registered to receive invalidates for |
| // create and updates operations. Get the actual region operation. |
| Part regionOpType = clientMessage.getPart(partCnt++); |
| Part numCqsPart = clientMessage.getPart(partCnt++); |
| if (isDebugEnabled) { |
| logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}", |
| numCqsPart.getInt() / 2); |
| } |
| partCnt = processCqs(clientMessage, partCnt, numCqsPart.getInt(), regionOpType.getInt(), |
| key, null); |
| this.isOpCompleted = true; |
| } |
| } catch (Exception e) { |
| final String message = |
| String.format( |
| "The following exception occurred while attempting to invalidate entry (region: %s key: %s)", |
| regionName, key); |
| handleException(message, e); |
| } |
| } |
| |
| /** |
| * locally destroy an entry |
| * |
| * @param clientMessage message describing the entry |
| */ |
| private void handleDestroy(Message clientMessage) { |
| String regionName = null; |
| Object key = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| this.isOpCompleted = false; |
| // Retrieve the data from the local-destroy message parts |
| if (isDebugEnabled) { |
| logger.debug("Received destroy message of length ({} bytes)", |
| clientMessage.getPayloadLength()); |
| } |
| |
| int partCnt = 0; |
| Part regionNamePart = clientMessage.getPart(partCnt++); |
| Part keyPart = clientMessage.getPart(partCnt++); |
| Part callbackArgumentPart = clientMessage.getPart(partCnt++); |
| |
| VersionTag versionTag = (VersionTag) clientMessage.getPart(partCnt++).getObject(); |
| if (versionTag != null) { |
| versionTag.replaceNullIDs((InternalDistributedMember) this.endpoint.getMemberId()); |
| } |
| |
| regionName = regionNamePart.getCachedString(); |
| key = keyPart.getStringOrObject(); |
| |
| Part isInterestListPassedPart = clientMessage.getPart(partCnt++); |
| Part hasCqsPart = clientMessage.getPart(partCnt++); |
| |
| boolean withInterest = ((Boolean) isInterestListPassedPart.getObject()).booleanValue(); |
| boolean withCQs = ((Boolean) hasCqsPart.getObject()).booleanValue(); |
| |
| Object callbackArgument = callbackArgumentPart.getObject(); |
| if (isDebugEnabled) { |
| logger.debug( |
| "Destroying entry for region: {} key: {} callbackArgument: {} withInterest={} withCQs={} version={}", |
| regionName, key, callbackArgument, withInterest, withCQs, versionTag); |
| } |
| |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| if (region == null) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug("Region named {} does not exist", regionName); |
| } |
| |
| } else if (region.hasServerProxy() && (withInterest || !withCQs)) { |
| EventID eventId = null; |
| try { |
| Part eid = clientMessage.getPart(clientMessage.getNumberOfParts() - 1); |
| eventId = (EventID) eid.getObject(); |
| |
| try { |
| region.basicBridgeClientDestroy(eventId.getDistributedMember(), key, callbackArgument, |
| this.qManager.getState().getProcessedMarker() || !this.isDurableClient, eventId, |
| versionTag); |
| } catch (ConcurrentCacheModificationException ignore) { |
| // allow CQs to be processed |
| } |
| |
| this.isOpCompleted = true; |
| if (isDebugEnabled) { |
| logger.debug("Destroyed entry for region: {} key: {} callbackArgument: {}", regionName, |
| key, callbackArgument); |
| } |
| } catch (EntryNotFoundException ignore) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug( |
| "Already destroyed entry for region: {} key: {} callbackArgument: {} eventId={}", |
| regionName, key, callbackArgument, eventId.expensiveToString()); |
| } |
| this.isOpCompleted = true; |
| } |
| } |
| |
| if (withCQs) { |
| Part numCqsPart = clientMessage.getPart(partCnt++); |
| if (isDebugEnabled) { |
| logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}", |
| numCqsPart.getInt() / 2); |
| } |
| partCnt = processCqs(clientMessage, partCnt, numCqsPart.getInt(), |
| clientMessage.getMessageType(), key, null); |
| this.isOpCompleted = true; |
| } |
| } catch (Exception e) { |
| String message = |
| String.format( |
| "The following exception occurred while attempting to destroy entry (region: %s key: %s)", |
| regionName, key); |
| handleException(message, e); |
| } |
| } |
| |
| /** |
| * Locally destroy a region |
| * |
| * @param clientMessage message describing the region |
| */ |
| private void handleDestroyRegion(Message clientMessage) { |
| String regionName = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| // Retrieve the data from the local-destroy-region message parts |
| if (isDebugEnabled) { |
| logger.debug("Received destroy region message of length ({} bytes)", |
| clientMessage.getPayloadLength()); |
| } |
| int partCnt = 0; |
| Part regionNamePart = clientMessage.getPart(partCnt++); |
| Part callbackArgumentPart = clientMessage.getPart(partCnt++); |
| regionName = regionNamePart.getCachedString(); |
| Object callbackArgument = callbackArgumentPart.getObject(); |
| |
| Part hasCqsPart = clientMessage.getPart(partCnt++); |
| |
| if (isDebugEnabled) { |
| logger.debug("Destroying region: {} callbackArgument: {}", regionName, callbackArgument); |
| } |
| |
| // Handle CQs if any on this region. |
| if ((Boolean) hasCqsPart.getObject()) { |
| Part numCqsPart = clientMessage.getPart(partCnt++); |
| if (isDebugEnabled) { |
| logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}", |
| numCqsPart.getInt() / 2); |
| } |
| // TODO: partCnt is unused -- does processCqs have side effects |
| partCnt = processCqs(clientMessage, partCnt, numCqsPart.getInt(), |
| clientMessage.getMessageType(), null, null); |
| } |
| |
| // Confirm that the region exists |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| if (region == null) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug("Region named {} does not exist", regionName); |
| } |
| return; |
| } |
| |
| // Verify that the region in question should respond to this message |
| if (region.hasServerProxy()) { |
| // Locally destroy the region |
| region.localDestroyRegion(callbackArgument); |
| |
| if (isDebugEnabled) { |
| logger.debug("Destroyed region: {} callbackArgument: {}", regionName, callbackArgument); |
| } |
| } |
| } catch (RegionDestroyedException ignore) { // already destroyed |
| if (isDebugEnabled) { |
| logger.debug("region already destroyed: {}", regionName); |
| } |
| } catch (Exception e) { |
| String message = |
| String.format("Caught an exception while attempting to destroy region %s", |
| regionName); |
| handleException(message, e); |
| } |
| } |
| |
| /** |
| * Locally clear a region |
| * |
| * @param clientMessage message describing the region to clear |
| */ |
| private void handleClearRegion(Message clientMessage) { |
| String regionName = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| // Retrieve the data from the clear-region message parts |
| if (isDebugEnabled) { |
| logger.debug("{}: Received clear region message of length ({} bytes)", this, |
| clientMessage.getPayloadLength()); |
| } |
| |
| int partCnt = 0; |
| Part regionNamePart = clientMessage.getPart(partCnt++); |
| Part callbackArgumentPart = clientMessage.getPart(partCnt++); |
| |
| Part hasCqsPart = clientMessage.getPart(partCnt++); |
| |
| regionName = regionNamePart.getCachedString(); |
| Object callbackArgument = callbackArgumentPart.getObject(); |
| if (isDebugEnabled) { |
| logger.debug("Clearing region: {} callbackArgument: {}", regionName, callbackArgument); |
| } |
| |
| if ((Boolean) hasCqsPart.getObject()) { |
| Part numCqsPart = clientMessage.getPart(partCnt++); |
| if (isDebugEnabled) { |
| logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}", |
| numCqsPart.getInt() / 2); |
| } |
| partCnt = processCqs(clientMessage, partCnt, numCqsPart.getInt(), |
| clientMessage.getMessageType(), null, null); |
| } |
| |
| // Confirm that the region exists |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| if (region == null) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug("Region named {} does not exist", regionName); |
| } |
| return; |
| } |
| |
| // Verify that the region in question should respond to this |
| // message |
| if (region.hasServerProxy()) { |
| // Locally clear the region |
| region.basicBridgeClientClear(callbackArgument, |
| this.qManager.getState().getProcessedMarker() || !this.isDurableClient); |
| |
| if (isDebugEnabled) { |
| logger.debug("Cleared region: {} callbackArgument: {}", regionName, callbackArgument); |
| } |
| } |
| } catch (Exception e) { |
| String message = |
| String.format("Caught the following exception while attempting to clear region %s", |
| regionName); |
| handleException(message, e); |
| } |
| } |
| |
| /** |
| * Locally invalidate a region NOTE: Added as part of bug#38048. The code only takes care of CQ |
| * processing. Support needs to be added for local region invalidate. |
| * |
| * @param clientMessage message describing the region to clear |
| */ |
| private void handleInvalidateRegion(Message clientMessage) { |
| String regionName = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| // Retrieve the data from the invalidate-region message parts |
| if (isDebugEnabled) { |
| logger.debug("{}: Received invalidate region message of length ({} bytes)", this, |
| clientMessage.getPayloadLength()); |
| } |
| |
| int partCnt = 0; |
| Part regionNamePart = clientMessage.getPart(partCnt++); |
| partCnt++; // Part callbackArgumentPart = m.getPart(partCnt++); |
| |
| Part hasCqsPart = clientMessage.getPart(partCnt++); |
| |
| regionName = regionNamePart.getCachedString(); |
| |
| if ((Boolean) hasCqsPart.getObject()) { |
| Part numCqsPart = clientMessage.getPart(partCnt++); |
| if (isDebugEnabled) { |
| logger.debug("Received message has CQ Event. Number of cqs interested in the event : {}", |
| numCqsPart.getInt() / 2); |
| } |
| // TODO: partCnt is unused |
| partCnt = processCqs(clientMessage, partCnt, numCqsPart.getInt(), |
| clientMessage.getMessageType(), null, null); |
| } |
| |
| // Confirm that the region exists |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| if (region == null) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug("Region named {} does not exist", regionName); |
| } |
| } |
| |
| } catch (Exception e) { |
| String message = |
| String.format("Caught the following exception while attempting to invalidate region %s.", |
| regionName); |
| handleException(message, e); |
| } |
| } |
| |
| /** |
| * Register instantiators locally |
| * |
| * @param clientMessage message describing the new instantiators |
| * @param eventId eventId of the instantiators |
| */ |
| private void handleRegisterInstantiator(Message clientMessage, EventID eventId) { |
| String instantiatorClassName = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| int noOfParts = clientMessage.getNumberOfParts(); |
| if (isDebugEnabled) { |
| logger.debug("{}: Received register instantiators message of parts {}", getName(), |
| noOfParts); |
| } |
| |
| Assert.assertTrue((noOfParts - 1) % 3 == 0); |
| for (int i = 0; i < noOfParts - 1; i += 3) { |
| instantiatorClassName = |
| (String) CacheServerHelper.deserialize(clientMessage.getPart(i).getSerializedForm()); |
| String instantiatedClassName = (String) CacheServerHelper |
| .deserialize(clientMessage.getPart(i + 1).getSerializedForm()); |
| int id = clientMessage.getPart(i + 2).getInt(); |
| InternalInstantiator.register(instantiatorClassName, instantiatedClassName, id, false, |
| eventId, null); |
| // distribute is false because we don't want to propagate this to servers recursively |
| } |
| |
| // CALLBACK TESTING PURPOSE ONLY |
| if (PoolImpl.IS_INSTANTIATOR_CALLBACK) { |
| ClientServerObserver clientServerObserver = ClientServerObserverHolder.getInstance(); |
| clientServerObserver.afterReceivingFromServer(eventId); |
| } |
| |
| } catch (Exception e) { |
| if (isDebugEnabled) { |
| logger.debug("{}: Caught following exception while attempting to read Instantiator : {}", |
| this, instantiatorClassName, e); |
| } |
| } |
| } |
| |
| private void handleRegisterDataSerializer(Message msg, EventID eventId) { |
| Class dataSerializerClass = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| int noOfParts = msg.getNumberOfParts(); |
| if (isDebugEnabled) { |
| logger.debug("{}: Received register dataserializer message of parts {}", getName(), |
| noOfParts); |
| } |
| |
| for (int i = 0; i < noOfParts - 1;) { |
| try { |
| String dataSerializerClassName = |
| (String) CacheServerHelper.deserialize(msg.getPart(i).getSerializedForm()); |
| int id = msg.getPart(i + 1).getInt(); |
| InternalDataSerializer.register(dataSerializerClassName, false, eventId, null, id); |
| // distribute is false because we don't want to propagate this to servers recursively |
| |
| int numOfClasses = msg.getPart(i + 2).getInt(); |
| int j = 0; |
| for (; j < numOfClasses; j++) { |
| String className = |
| (String) CacheServerHelper.deserialize(msg.getPart(i + 3 + j).getSerializedForm()); |
| InternalDataSerializer.updateSupportedClassesMap(dataSerializerClassName, className); |
| } |
| |
| i += 3 + j; |
| } catch (ClassNotFoundException e) { |
| if (isDebugEnabled) { |
| logger.debug( |
| "{}: Caught following exception while attempting to read DataSerializer : {}", this, |
| dataSerializerClass, e); |
| } |
| } |
| } |
| |
| // CALLBACK TESTING PURPOSE ONLY |
| if (PoolImpl.IS_INSTANTIATOR_CALLBACK) { |
| ClientServerObserver bo = ClientServerObserverHolder.getInstance(); |
| bo.afterReceivingFromServer(eventId); |
| } |
| |
| } catch (Exception e) { |
| if (isDebugEnabled) { |
| logger.debug("{}: Caught following exception while attempting to read DataSerializer : {}", |
| this, dataSerializerClass, e); |
| } |
| } |
| } |
| |
| /** |
| * Processes message to invoke CQ listeners. |
| */ |
| private int processCqs(Message clientMessage, int startMessagePart, int numCqParts, |
| int messageType, Object key, Object value) { |
| return processCqs(clientMessage, startMessagePart, numCqParts, messageType, key, value, null, |
| null); |
| } |
| |
| private int processCqs(Message clientMessage, int startMessagePart, int numCqParts, |
| int messageType, Object key, Object value, byte[] delta, EventID eventId) { |
| HashMap cqs = new HashMap(); |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| for (int cqCnt = 0; cqCnt < numCqParts;) { |
| StringBuilder sb = null; |
| if (isDebugEnabled) { |
| sb = new StringBuilder(100); |
| sb.append("found these queries: "); |
| } |
| try { |
| // Get CQ Name. |
| Part cqNamePart = clientMessage.getPart(startMessagePart + cqCnt++); |
| // Get CQ Op. |
| Part cqOpPart = clientMessage.getPart(startMessagePart + cqCnt++); |
| cqs.put(cqNamePart.getString(), cqOpPart.getInt()); |
| |
| if (sb != null) { |
| sb.append(cqNamePart.getString()).append(" op=").append(cqOpPart.getInt()).append(" "); |
| } |
| } catch (Exception ignore) { |
| logger.warn( |
| "Error while processing the CQ Message. Problem with reading message for CQ# : {}", |
| cqCnt); |
| } |
| if (isDebugEnabled) { |
| logger.debug(sb); |
| } |
| } |
| |
| CqService cqService = this.cache.getCqService(); |
| try { |
| cqService.dispatchCqListeners(cqs, messageType, key, value, delta, this.qManager, eventId); |
| } catch (Exception ex) { |
| logger.warn("Failed to invoke CQ Dispatcher. Error : {}", |
| ex.getMessage()); |
| if (isDebugEnabled) { |
| logger.debug("Failed to invoke CQ Dispatcher.", ex); |
| } |
| } |
| |
| return startMessagePart + numCqParts; |
| } |
| |
| private void handleRegisterInterest(Message clientMessage) { |
| String regionName = null; |
| Object key = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| // Retrieve the data from the add interest message parts |
| if (isDebugEnabled) { |
| logger.debug("{}: Received add interest message of length ({} bytes)", this, |
| clientMessage.getPayloadLength()); |
| } |
| |
| int partCnt = 0; |
| Part regionNamePart = clientMessage.getPart(partCnt++); |
| Part keyPart = clientMessage.getPart(partCnt++); |
| Part interestTypePart = clientMessage.getPart(partCnt++); |
| Part interestResultPolicyPart = clientMessage.getPart(partCnt++); |
| Part isDurablePart = clientMessage.getPart(partCnt++); |
| Part receiveUpdatesAsInvalidatesPart = clientMessage.getPart(partCnt++); |
| |
| regionName = regionNamePart.getCachedString(); |
| key = keyPart.getStringOrObject(); |
| int interestType = (Integer) interestTypePart.getObject(); |
| byte interestResultPolicy = (Byte) interestResultPolicyPart.getObject(); |
| boolean isDurable = (Boolean) isDurablePart.getObject(); |
| boolean receiveUpdatesAsInvalidates = (Boolean) receiveUpdatesAsInvalidatesPart.getObject(); |
| |
| // Confirm that region exists |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| if (region == null) { |
| if (isDebugEnabled && !quitting()) { |
| logger.debug("{}: Region named {} does not exist", this, regionName); |
| } |
| return; |
| } |
| |
| // Verify that the region in question should respond to this message |
| if (!region.hasServerProxy()) { |
| return; |
| } |
| |
| if (key instanceof List) { |
| region.getServerProxy().addListInterest((List) key, |
| InterestResultPolicy.fromOrdinal(interestResultPolicy), isDurable, |
| receiveUpdatesAsInvalidates); |
| } else { |
| region.getServerProxy().addSingleInterest(key, interestType, |
| InterestResultPolicy.fromOrdinal(interestResultPolicy), isDurable, |
| receiveUpdatesAsInvalidates); |
| } |
| } catch (Exception e) { |
| String message = |
| ": The following exception occurred while attempting to add interest (region: " |
| + regionName + " key: " + key + "): "; |
| handleException(message, e); |
| } |
| } |
| |
| private void handleUnregisterInterest(Message clientMessage) { |
| String regionName = null; |
| Object key = null; |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| try { |
| // Retrieve the data from the remove interest message parts |
| if (isDebugEnabled) { |
| logger.debug("{}: Received remove interest message of length ({} bytes)", this, |
| clientMessage.getPayloadLength()); |
| } |
| |
| int partCnt = 0; |
| Part regionNamePart = clientMessage.getPart(partCnt++); |
| Part keyPart = clientMessage.getPart(partCnt++); |
| Part interestTypePart = clientMessage.getPart(partCnt++); |
| Part isDurablePart = clientMessage.getPart(partCnt++); |
| Part receiveUpdatesAsInvalidatesPart = clientMessage.getPart(partCnt++); |
| // Not reading the eventId part |
| |
| regionName = regionNamePart.getCachedString(); |
| key = keyPart.getStringOrObject(); |
| int interestType = (Integer) interestTypePart.getObject(); |
| boolean isDurable = (Boolean) isDurablePart.getObject(); |
| boolean receiveUpdatesAsInvalidates = (Boolean) receiveUpdatesAsInvalidatesPart.getObject(); |
| |
| // Confirm that region exists |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| if (region == null) { |
| if (isDebugEnabled) { |
| logger.debug("{}: Region named {} does not exist", this, regionName); |
| } |
| return; |
| } |
| |
| // Verify that the region in question should respond to this message |
| if (!region.hasServerProxy()) { |
| return; |
| } |
| |
| if (key instanceof List) { |
| region.getServerProxy().removeListInterest((List) key, isDurable, |
| receiveUpdatesAsInvalidates); |
| } else { |
| region.getServerProxy().removeSingleInterest(key, interestType, isDurable, |
| receiveUpdatesAsInvalidates); |
| } |
| } catch (Exception e) { |
| String message = |
| ": The following exception occurred while attempting to add interest (region: " |
| + regionName + " key: " + key + "): "; |
| handleException(message, e); |
| } |
| } |
| |
| private void handleTombstoneOperation(Message clientMessage) { |
| String regionName = "unknown"; |
| |
| try { // not sure why this isn't done by the caller |
| int partIdx = 0; |
| |
| // see ClientTombstoneMessage.getGFE70Message |
| regionName = clientMessage.getPart(partIdx++).getCachedString(); |
| int op = clientMessage.getPart(partIdx++).getInt(); |
| LocalRegion region = (LocalRegion) this.cacheHelper.getRegion(regionName); |
| |
| if (region == null) { |
| if (!quitting()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Region named {} does not exist", this, regionName); |
| } |
| } |
| return; |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Received tombstone operation for region {} with operation={}", this, |
| region, op); |
| } |
| |
| if (!region.getConcurrencyChecksEnabled()) { |
| return; |
| } |
| |
| switch (op) { |
| case 0: |
| Map<VersionSource, Long> regionGCVersions = |
| (Map<VersionSource, Long>) clientMessage.getPart(partIdx++).getObject(); |
| EventID eventID = (EventID) clientMessage.getPart(partIdx++).getObject(); |
| region.expireTombstones(regionGCVersions, eventID, null); |
| break; |
| |
| case 1: |
| Set<Object> removedKeys = (Set<Object>) clientMessage.getPart(partIdx++).getObject(); |
| region.expireTombstoneKeys(removedKeys); |
| break; |
| |
| default: |
| throw new IllegalArgumentException("unknown operation type " + op); |
| } |
| } catch (Exception e) { |
| handleException(": exception while removing tombstones from " + regionName, e); |
| } |
| } |
| |
| /** |
| * Indicate whether the updater or the system is trying to terminate |
| * |
| * @return true if we are trying to stop |
| */ |
| private boolean quitting() { |
| if (isInterrupted()) { |
| // Any time an interrupt is thrown at this thread, regard it as a request to terminate |
| return true; |
| } |
| if (!this.continueProcessing.get()) { |
| // de facto flag indicating we are to stop |
| return true; |
| } |
| if (this.cache != null && this.cache.getCancelCriterion().isCancelInProgress()) { |
| // System is cancelling |
| return true; |
| } |
| |
| // The pool stuff is really sick, so it's possible for us to have a distributed |
| // system that is not the same as our cache. Check it just in case... |
| if (this.system.getCancelCriterion().isCancelInProgress()) { |
| return true; |
| } |
| |
| // All clear on this end, boss. |
| return false; |
| } |
| |
| private void waitForFailedUpdater() { |
| boolean gotInterrupted = false; |
| try { |
| if (this.failedUpdater != null) { |
| logger.info("{} is waiting for {} to complete.", |
| new Object[] {this, this.failedUpdater}); |
| while (this.failedUpdater.isAlive()) { |
| if (quitting()) { |
| return; |
| } |
| this.failedUpdater.join(5000); |
| } |
| } |
| } catch (InterruptedException ignore) { |
| gotInterrupted = true; |
| // just bail, because I have not done anything yet |
| } finally { |
| if (!gotInterrupted && this.failedUpdater != null) { |
| logger.info("{} has completed waiting for {}", |
| new Object[] {this, this.failedUpdater}); |
| this.failedUpdater = null; |
| } |
| } |
| } |
| |
| /** |
| * Processes messages received from the server. |
| * |
| * Only certain types of messages are handled. |
| * |
| * TODO: Method 'processMessages' is too complex to analyze by data flow algorithm |
| * |
| * @see MessageType#CLIENT_MARKER |
| * @see MessageType#LOCAL_CREATE |
| * @see MessageType#LOCAL_UPDATE |
| * @see MessageType#LOCAL_INVALIDATE |
| * @see MessageType#LOCAL_DESTROY |
| * @see MessageType#LOCAL_DESTROY_REGION |
| * @see MessageType#CLEAR_REGION |
| * @see ClientUpdateMessage |
| */ |
| private void processMessages() { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| final int headerReadTimeout = (int) Math.round(serverQueueStatus.getPingInterval() |
| * qManager.getPool().getSubscriptionTimeoutMultiplier() * 1.25); |
| |
| try { |
| Message clientMessage = initializeMessage(); |
| |
| if (quitting()) { |
| if (isDebugEnabled) { |
| logger.debug("processMessages quitting early because we have stopped"); |
| } |
| // our caller calls close which will notify all waiters for our init |
| return; |
| } |
| |
| logger.info("{} : ready to process messages.", this); |
| |
| while (this.continueProcessing.get()) { |
| if (quitting()) { |
| if (isDebugEnabled) { |
| logger.debug("termination detected"); |
| } |
| // our caller calls close which will notify all waiters for our init |
| return; |
| } |
| |
| // the endpoint died while this thread was sleeping. |
| if (this.endpoint.isClosed()) { |
| if (isDebugEnabled) { |
| logger.debug("endpoint died"); |
| } |
| this.continueProcessing.set(false);// = false; |
| break; |
| } |
| |
| try { |
| // Read the message |
| clientMessage.receiveWithHeaderReadTimeout(headerReadTimeout); |
| |
| // Wait for the previously failed cache client updater |
| // to finish. This will avoid out of order messages. |
| waitForFailedUpdater(); |
| this.cache.waitForRegisterInterestsInProgress(); |
| if (quitting()) { |
| if (isDebugEnabled) { |
| logger.debug("processMessages quitting before processing message"); |
| } |
| break; |
| } |
| |
| // If the message is a ping, ignore it |
| if (clientMessage.getMessageType() == MessageType.SERVER_TO_CLIENT_PING) { |
| if (isDebugEnabled) { |
| logger.debug("{}: Received ping", this); |
| } |
| continue; |
| } |
| |
| boolean isDeltaSent = false; |
| boolean isCreateOrUpdate = clientMessage.getMessageType() == MessageType.LOCAL_CREATE |
| || clientMessage.getMessageType() == MessageType.LOCAL_UPDATE; |
| if (isCreateOrUpdate) { |
| isDeltaSent = (Boolean) clientMessage.getPart(2).getObject(); |
| } |
| |
| // extract the eventId and verify if it is a duplicate event |
| // if it is a duplicate event, ignore |
| // @since GemFire 5.1 |
| int numberOfParts = clientMessage.getNumberOfParts(); |
| Part eid = clientMessage.getPart(numberOfParts - 1); |
| |
| // TODO the message handling methods also deserialized the eventID - inefficient |
| EventID eventId = (EventID) eid.getObject(); |
| |
| // no need to verify if the instantiator msg is duplicate or not |
| if (clientMessage.getMessageType() != MessageType.REGISTER_INSTANTIATORS |
| && clientMessage.getMessageType() != MessageType.REGISTER_DATASERIALIZERS) { |
| if (this.qManager.getState().verifyIfDuplicate(eventId, |
| !(this.isDurableClient || isDeltaSent))) { |
| continue; |
| } |
| } |
| |
| if (logger.isTraceEnabled(LogMarker.BRIDGE_SERVER_VERBOSE)) { |
| logger.trace(LogMarker.BRIDGE_SERVER_VERBOSE, "Processing event with id {}", |
| eventId.expensiveToString()); |
| } |
| |
| this.isOpCompleted = true; |
| |
| // Process the message |
| switch (clientMessage.getMessageType()) { |
| case MessageType.LOCAL_CREATE: |
| case MessageType.LOCAL_UPDATE: |
| handleUpdate(clientMessage); |
| break; |
| case MessageType.LOCAL_INVALIDATE: |
| handleInvalidate(clientMessage); |
| break; |
| case MessageType.LOCAL_DESTROY: |
| handleDestroy(clientMessage); |
| break; |
| case MessageType.LOCAL_DESTROY_REGION: |
| handleDestroyRegion(clientMessage); |
| break; |
| case MessageType.CLEAR_REGION: |
| handleClearRegion(clientMessage); |
| break; |
| case MessageType.REGISTER_INSTANTIATORS: |
| handleRegisterInstantiator(clientMessage, eventId); |
| break; |
| case MessageType.REGISTER_DATASERIALIZERS: |
| handleRegisterDataSerializer(clientMessage, eventId); |
| break; |
| case MessageType.CLIENT_MARKER: |
| handleMarker(clientMessage); |
| break; |
| case MessageType.INVALIDATE_REGION: |
| handleInvalidateRegion(clientMessage); |
| break; |
| case MessageType.CLIENT_REGISTER_INTEREST: |
| handleRegisterInterest(clientMessage); |
| break; |
| case MessageType.CLIENT_UNREGISTER_INTEREST: |
| handleUnregisterInterest(clientMessage); |
| break; |
| case MessageType.TOMBSTONE_OPERATION: |
| handleTombstoneOperation(clientMessage); |
| break; |
| default: |
| logger.warn("{}: Received an unsupported message (type={})", |
| new Object[] {this, MessageType.getString(clientMessage.getMessageType())}); |
| break; |
| } |
| |
| if (this.isOpCompleted && (this.isDurableClient || isDeltaSent)) { |
| this.qManager.getState().verifyIfDuplicate(eventId, true); |
| } |
| |
| // TODO we should maintain the client's "live" view of the server |
| // but we don't because the server health monitor needs traffic |
| // originating from the client |
| // and by updating the last update stat, the ServerMonitor is less |
| // likely to send pings... |
| // and the ClientHealthMonitor will cause a disconnect |
| |
| } catch (InterruptedIOException ignore) { |
| // Per Sun's support web site, this exception seems to be peculiar |
| // to Solaris, and may eventually not even be generated there. |
| // |
| // When this exception is thrown, the thread has been interrupted, but |
| // isInterrupted() is false. (How very odd!) |
| // |
| // We regard it the same as an InterruptedException |
| |
| this.continueProcessing.set(false); |
| if (isDebugEnabled) { |
| logger.debug("InterruptedIOException"); |
| } |
| |
| } catch (IOException e) { |
| // Either the server went away, or we caught a closing condition. |
| if (!quitting()) { |
| // Server departed; print a message. |
| ClientServerObserver clientServerObserver = ClientServerObserverHolder.getInstance(); |
| clientServerObserver.beforeFailoverByCacheClientUpdater(this.location); |
| this.eManager.serverCrashed(this.endpoint); |
| if (isDebugEnabled) { |
| logger.debug("Caught the following exception and will exit", e); |
| } |
| } // !quitting |
| |
| // In any event, terminate this thread. |
| this.continueProcessing.set(false); |
| if (isDebugEnabled) { |
| logger.debug("terminated due to IOException"); |
| } |
| |
| } catch (Exception e) { |
| if (!quitting()) { |
| ClientServerObserver clientServerObserver = ClientServerObserverHolder.getInstance(); |
| clientServerObserver.beforeFailoverByCacheClientUpdater(this.location); |
| this.eManager.serverCrashed(this.endpoint); |
| String message = ": Caught the following exception and will exit: "; |
| handleException(message, e); |
| } |
| |
| // In any event, terminate this thread. |
| this.continueProcessing.set(false);// = false; // force termination |
| if (isDebugEnabled) { |
| logger.debug("CCU terminated due to Exception"); |
| } |
| |
| } finally { |
| clientMessage.clear(); |
| } |
| } // while |
| |
| } finally { |
| if (isDebugEnabled) { |
| logger.debug("has stopped and cleaning the helper .."); |
| } |
| close(); // added to fix some race conditions associated with 38382 |
| // this will make sure that if this thread dies without starting QueueMgr then it will start.. |
| // 1. above we ignore InterruptedIOException and this thread dies without informing QueueMgr |
| // 2. if there is some other race condition with continueProcessing flag |
| this.qManager.checkEndpoint(this, this.endpoint); |
| } |
| } |
| |
| /** |
| * Conditionally print a warning describing the failure |
| * <p> |
| * Signals run thread to stop. Messages are not printed if the thread or the distributed system |
| * has already been instructed to terminate. |
| * |
| * @param message contextual string for the failure |
| * @param exception underlying exception |
| */ |
| private void handleException(String message, Exception exception) { |
| boolean unexpected = !quitting(); |
| |
| // If this was a surprise, print a warning. |
| if (unexpected && !(exception instanceof CancelException)) { |
| logger.warn(String.format("%s : %s : %s", |
| new Object[] {this, message, exception}), exception); |
| } |
| // We can't shutdown the client updater just because of an exception. |
| // Let the caller decide if we should continue running or not. |
| } |
| |
| /** |
| * Return an object from serialization. Only used in debug logging. |
| * |
| * @param serializedBytes the serialized form |
| * @return the deserialized object |
| */ |
| private Object deserialize(byte[] serializedBytes) { |
| Object deserializedObject = serializedBytes; |
| // This is a debugging method so ignore all exceptions like ClassNotFoundException |
| try { |
| ByteArrayDataInput dis = new ByteArrayDataInput(serializedBytes); |
| deserializedObject = DataSerializer.readObject(dis); |
| } catch (ClassNotFoundException | IOException ignore) { |
| } |
| return deserializedObject; |
| } |
| |
| /** |
| * @return the local port of our {@link #socket} |
| */ |
| protected int getLocalPort() { |
| return this.socket.getLocalPort(); |
| } |
| |
| @Override |
| public void onDisconnect(InternalDistributedSystem sys) { |
| stopUpdater(); |
| } |
| |
| private void verifySocketBufferSize(int requestedBufferSize, int actualBufferSize, String type) { |
| if (actualBufferSize < requestedBufferSize) { |
| logger.info("Socket {} is {} instead of the requested {}.", |
| new Object[] {type + " buffer size", actualBufferSize, requestedBufferSize}); |
| } |
| } |
| |
| public static class StatisticsProvider { |
| public CCUStats createStatistics(DistributedSystem system, ServerLocation location) { |
| return new CCUStats(system, location); |
| } |
| } |
| |
| /** |
| * Stats for a CacheClientUpdater. Currently the only thing measured are incoming bytes on the |
| * wire |
| * |
| * @since GemFire 5.7 |
| */ |
| public static class CCUStats implements MessageStats { |
| |
| @Immutable |
| private static final StatisticsType type; |
| private static final int messagesBeingReceivedId; |
| private static final int messageBytesBeingReceivedId; |
| private static final int receivedBytesId; |
| |
| static { |
| StatisticsTypeFactory f = StatisticsTypeFactoryImpl.singleton(); |
| type = f.createType("CacheClientUpdaterStats", "Statistics about incoming subscription data", |
| new StatisticDescriptor[] { |
| f.createLongCounter("receivedBytes", |
| "Total number of bytes received from the server.", "bytes"), |
| f.createIntGauge("messagesBeingReceived", |
| "Current number of message being received off the network or being processed after reception.", |
| "messages"), |
| f.createLongGauge("messageBytesBeingReceived", |
| "Current number of bytes consumed by messages being received or processed.", |
| "bytes"),}); |
| receivedBytesId = type.nameToId("receivedBytes"); |
| messagesBeingReceivedId = type.nameToId("messagesBeingReceived"); |
| messageBytesBeingReceivedId = type.nameToId("messageBytesBeingReceived"); |
| } |
| |
| // instance fields |
| private final Statistics stats; |
| |
| CCUStats(DistributedSystem ids, ServerLocation location) { |
| // no need for atomic since only a single thread will be writing these |
| this.stats = ids.createStatistics(type, "CacheClientUpdater-" + location); |
| } |
| |
| public void close() { |
| this.stats.close(); |
| } |
| |
| @Override |
| public void incReceivedBytes(long v) { |
| this.stats.incLong(receivedBytesId, v); |
| } |
| |
| @Override |
| public void incSentBytes(long v) { |
| // noop since we never send messages |
| } |
| |
| @Override |
| public void incMessagesBeingReceived(int bytes) { |
| this.stats.incInt(messagesBeingReceivedId, 1); |
| if (bytes > 0) { |
| this.stats.incLong(messageBytesBeingReceivedId, bytes); |
| } |
| } |
| |
| @Override |
| public void decMessagesBeingReceived(int bytes) { |
| this.stats.incInt(messagesBeingReceivedId, -1); |
| if (bytes > 0) { |
| this.stats.incLong(messageBytesBeingReceivedId, -bytes); |
| } |
| } |
| |
| /** |
| * Returns the current time (ns). |
| * |
| * @return the current time (ns) |
| */ |
| public long startTime() { |
| return DistributionStats.getStatTime(); |
| } |
| } |
| |
| @Override |
| public boolean isProcessing() { |
| return this.continueProcessing.get(); |
| } |
| } |
