| /* |
| * 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.wan.parallel; |
| |
| import static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.Comparator; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Random; |
| import java.util.Set; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.LinkedBlockingQueue; |
| import java.util.concurrent.TimeUnit; |
| |
| import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet; |
| import org.apache.logging.log4j.Logger; |
| |
| import org.apache.geode.CancelException; |
| import org.apache.geode.SystemFailure; |
| import org.apache.geode.annotations.internal.MutableForTesting; |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.AttributesMutator; |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.CacheListener; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.EntryNotFoundException; |
| import org.apache.geode.cache.EvictionAction; |
| import org.apache.geode.cache.EvictionAttributes; |
| import org.apache.geode.cache.PartitionAttributesFactory; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionAttributes; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.cache.AbstractBucketRegionQueue; |
| import org.apache.geode.internal.cache.BucketNotFoundException; |
| import org.apache.geode.internal.cache.BucketRegion; |
| import org.apache.geode.internal.cache.BucketRegionQueue; |
| import org.apache.geode.internal.cache.ColocationHelper; |
| import org.apache.geode.internal.cache.Conflatable; |
| import org.apache.geode.internal.cache.DiskRegionStats; |
| import org.apache.geode.internal.cache.DistributedRegion; |
| import org.apache.geode.internal.cache.EntryEventImpl; |
| import org.apache.geode.internal.cache.ForceReattemptException; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.InternalRegionArguments; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.LocalRegion.InitializationLevel; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.PartitionedRegionHelper; |
| import org.apache.geode.internal.cache.PrimaryBucketException; |
| import org.apache.geode.internal.cache.RegionQueue; |
| import org.apache.geode.internal.cache.wan.AbstractGatewaySender; |
| import org.apache.geode.internal.cache.wan.AsyncEventQueueConfigurationException; |
| import org.apache.geode.internal.cache.wan.GatewaySenderConfigurationException; |
| import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl; |
| import org.apache.geode.internal.cache.wan.GatewaySenderException; |
| import org.apache.geode.internal.cache.wan.GatewaySenderStats; |
| import org.apache.geode.internal.size.SingleObjectSizer; |
| import org.apache.geode.internal.statistics.StatisticsClock; |
| import org.apache.geode.internal.util.concurrent.StoppableCondition; |
| import org.apache.geode.internal.util.concurrent.StoppableReentrantLock; |
| import org.apache.geode.logging.internal.executors.LoggingExecutors; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| import org.apache.geode.management.ManagementService; |
| import org.apache.geode.management.internal.beans.AsyncEventQueueMBean; |
| import org.apache.geode.management.internal.beans.GatewaySenderMBean; |
| |
| public class ParallelGatewaySenderQueue implements RegionQueue { |
| |
| protected static final Logger logger = LogService.getLogger(); |
| |
| protected final Map<String, PartitionedRegion> userRegionNameToshadowPRMap = |
| new ConcurrentHashMap<String, PartitionedRegion>(); |
| |
| // <PartitionedRegion, Map<Integer, List<Object>>> |
| private final Map regionToDispatchedKeysMap = new ConcurrentHashMap(); |
| |
| protected final StoppableReentrantLock buckToDispatchLock; |
| |
| private final StoppableCondition regionToDispatchedKeysMapEmpty; |
| |
| protected final StoppableReentrantLock queueEmptyLock; |
| |
| private volatile boolean isQueueEmpty = true; |
| |
| /** |
| * False signal is fine on this condition. As processor will loop again and find out if it was a |
| * false signal. However, make sure that whatever scenario can cause an entry to be peeked shoudld |
| * signal the processor to unblock. |
| */ |
| private StoppableCondition queueEmptyCondition; |
| |
| protected final GatewaySenderStats stats; |
| |
| protected volatile boolean resetLastPeeked = false; |
| |
| /** |
| * There will be one shadow pr for each of the the PartitionedRegion which has added the |
| * GatewaySender Fix for Bug#45917 We maintain a tempQueue to queue events when buckets are not |
| * available locally. |
| */ |
| private final ConcurrentMap<Integer, BlockingQueue<GatewaySenderEventImpl>> bucketToTempQueueMap = |
| new ConcurrentHashMap<Integer, BlockingQueue<GatewaySenderEventImpl>>(); |
| |
| /** |
| * The default frequency (in milliseconds) at which a message will be sent by the primary to all |
| * the secondary nodes to remove the events which have already been dispatched from the queue. |
| */ |
| public static final int DEFAULT_MESSAGE_SYNC_INTERVAL = 10; |
| |
| // TODO:REF: how to change the message sync interval ? should it be common for serial and parallel |
| @MutableForTesting |
| protected static volatile int messageSyncInterval = DEFAULT_MESSAGE_SYNC_INTERVAL; |
| |
| // TODO:REF: name change for thread, as it appears in the log |
| private BatchRemovalThread removalThread = null; |
| |
| protected BlockingQueue<GatewaySenderEventImpl> peekedEvents = |
| new LinkedBlockingQueue<GatewaySenderEventImpl>(); |
| |
| /** |
| * The peekedEventsProcessing queue is used when the batch size is reduced due to a |
| * MessageTooLargeException |
| */ |
| private BlockingQueue<GatewaySenderEventImpl> peekedEventsProcessing = |
| new LinkedBlockingQueue<GatewaySenderEventImpl>(); |
| |
| /** |
| * The peekedEventsProcessingInProgress boolean denotes that processing existing peeked events is |
| * in progress |
| */ |
| private boolean peekedEventsProcessingInProgress = false; |
| |
| public final AbstractGatewaySender sender; |
| |
| public static final int WAIT_CYCLE_SHADOW_BUCKET_LOAD = 10; |
| |
| public static final String QSTRING = "_PARALLEL_GATEWAY_SENDER_QUEUE"; |
| |
| /** |
| * Fixed size Thread pool for conflating the events in the queue. The size of the thread pool is |
| * set to the number of processors available to the JVM. There will be one thread pool per |
| * ParallelGatewaySender on a node. |
| */ |
| private ExecutorService conflationExecutor; |
| |
| /** |
| * This class carries out the actual removal of the previousTailKey from QPR. The class implements |
| * Runnable and the destroy operation is done in the run method. The Runnable is executed by the |
| * one of the threads in the conflation thread pool configured above. |
| */ |
| private class ConflationHandler implements Runnable { |
| Conflatable conflatableObject; |
| |
| Long previousTailKeyTobeRemoved; |
| |
| int bucketId; |
| |
| public ConflationHandler(Conflatable conflatableObject, int bId, Long previousTailKey) { |
| this.conflatableObject = conflatableObject; |
| this.previousTailKeyTobeRemoved = previousTailKey; |
| this.bucketId = bId; |
| } |
| |
| @Override |
| public void run() { |
| PartitionedRegion prQ = null; |
| GatewaySenderEventImpl event = (GatewaySenderEventImpl) conflatableObject; |
| try { |
| String regionPath = |
| ColocationHelper.getLeaderRegion((PartitionedRegion) event.getRegion()).getFullPath(); |
| prQ = userRegionNameToshadowPRMap.get(regionPath); |
| destroyEventFromQueue(prQ, bucketId, previousTailKeyTobeRemoved); |
| } catch (EntryNotFoundException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Not conflating {} due to EntryNotFoundException", this, |
| conflatableObject.getKeyToConflate()); |
| } |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Conflated {} for key={} in queue for region={}", this, |
| conflatableObject.getValueToConflate(), conflatableObject.getKeyToConflate(), |
| prQ.getName()); |
| } |
| } |
| |
| private Object deserialize(Object serializedBytes) { |
| Object deserializedObject = serializedBytes; |
| if (serializedBytes instanceof byte[]) { |
| byte[] serializedBytesCast = (byte[]) serializedBytes; |
| // This is a debugging method so ignore all exceptions like |
| // ClassNotFoundException |
| try { |
| deserializedObject = EntryEventImpl.deserialize(serializedBytesCast); |
| } catch (Exception e) { |
| } |
| } |
| return deserializedObject; |
| } |
| } |
| |
| protected final int index; |
| |
| protected final int nDispatcher; |
| |
| private MetaRegionFactory metaRegionFactory; |
| |
| public ParallelGatewaySenderQueue(AbstractGatewaySender sender, Set<Region> userRegions, int idx, |
| int nDispatcher) { |
| this(sender, userRegions, idx, nDispatcher, new MetaRegionFactory()); |
| } |
| |
| ParallelGatewaySenderQueue(AbstractGatewaySender sender, Set<Region> userRegions, int idx, |
| int nDispatcher, MetaRegionFactory metaRegionFactory) { |
| |
| this.metaRegionFactory = metaRegionFactory; |
| |
| this.index = idx; |
| this.nDispatcher = nDispatcher; |
| this.stats = sender.getStatistics(); |
| this.sender = sender; |
| |
| List<Region> listOfRegions = new ArrayList<Region>(userRegions); |
| Collections.sort(listOfRegions, new Comparator<Region>() { |
| @Override |
| public int compare(Region o1, Region o2) { |
| return o1.getFullPath().compareTo(o2.getFullPath()); |
| } |
| }); |
| |
| for (Region userRegion : listOfRegions) { |
| if (userRegion instanceof PartitionedRegion) { |
| addShadowPartitionedRegionForUserPR((PartitionedRegion) userRegion); |
| } else { |
| // Fix for Bug#51491. Once decided to support this configuration we have call |
| // addShadowPartitionedRegionForUserRR |
| if (this.sender.getId().contains(AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX)) { |
| throw new AsyncEventQueueConfigurationException( |
| String.format( |
| "Parallel Async Event Queue %s can not be used with replicated region %s", |
| new Object[] { |
| AsyncEventQueueImpl.getAsyncEventQueueIdFromSenderId(this.sender.getId()), |
| userRegion.getFullPath()})); |
| } |
| throw new GatewaySenderConfigurationException( |
| String.format("Parallel gateway sender %s can not be used with replicated region %s", |
| new Object[] {this.sender.getId(), userRegion.getFullPath()})); |
| } |
| } |
| |
| buckToDispatchLock = new StoppableReentrantLock(sender.getCancelCriterion()); |
| regionToDispatchedKeysMapEmpty = buckToDispatchLock.newCondition(); |
| |
| queueEmptyLock = new StoppableReentrantLock(sender.getCancelCriterion()); |
| queueEmptyCondition = queueEmptyLock.newCondition(); |
| |
| // initialize the conflation thread pool if conflation is enabled |
| if (sender.isBatchConflationEnabled()) { |
| initializeConflationThreadPool(); |
| } |
| } |
| |
| /** Start the background batch removal thread. */ |
| public void start() { |
| // at present, this won't be accessed by multiple threads, |
| // still, it is safer approach to synchronize it |
| synchronized (ParallelGatewaySenderQueue.class) { |
| if (removalThread == null) { |
| removalThread = new BatchRemovalThread(this.sender.getCache(), this); |
| removalThread.start(); |
| } |
| } |
| } |
| |
| public void addShadowPartitionedRegionForUserRR(DistributedRegion userRegion) { |
| this.sender.getLifeCycleLock().writeLock().lock(); |
| PartitionedRegion prQ = null; |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "addShadowPartitionedRegionForUserRR: Going to create shadowpr for userRegion {}", |
| userRegion.getFullPath()); |
| } |
| |
| try { |
| String regionName = userRegion.getFullPath(); |
| |
| if (this.userRegionNameToshadowPRMap.containsKey(regionName)) |
| return; |
| |
| InternalCache cache = sender.getCache(); |
| final String prQName = getQueueName(sender.getId(), userRegion.getFullPath()); |
| prQ = (PartitionedRegion) cache.getRegion(prQName); |
| if (prQ == null) { |
| // TODO:REF:Avoid deprecated apis |
| AttributesFactory fact = new AttributesFactory(); |
| // Fix for 48621 - don't enable concurrency checks |
| // for queue buckets., event with persistence |
| fact.setConcurrencyChecksEnabled(false); |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(sender.getMaxParallelismForReplicatedRegion()); |
| int localMaxMemory = |
| userRegion.getDataPolicy().withStorage() ? sender.getMaximumQueueMemory() : 0; |
| pfact.setLocalMaxMemory(localMaxMemory); |
| pfact.setRedundantCopies(3); // TODO:Kishor : THis need to be handled nicely |
| pfact.setPartitionResolver(new RREventIDResolver()); |
| if (sender.isPersistenceEnabled()) { |
| fact.setDataPolicy(DataPolicy.PERSISTENT_PARTITION); |
| } |
| |
| fact.setDiskStoreName(sender.getDiskStoreName()); |
| |
| // if persistence is enabled, set the diskSyncronous to whatever user |
| // has set |
| // else set it to false |
| // optimize with above check of enable persistence |
| if (sender.isPersistenceEnabled()) |
| fact.setDiskSynchronous(sender.isDiskSynchronous()); |
| else { |
| fact.setDiskSynchronous(false); |
| } |
| |
| // allow for no overflow directory |
| EvictionAttributes ea = EvictionAttributes.createLIFOMemoryAttributes( |
| sender.getMaximumQueueMemory(), EvictionAction.OVERFLOW_TO_DISK); |
| |
| fact.setEvictionAttributes(ea); |
| fact.setPartitionAttributes(pfact.create()); |
| |
| final RegionAttributes ra = fact.create(); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Attempting to create queue region: {}", this, prQName); |
| } |
| |
| ParallelGatewaySenderQueueMetaRegion meta = |
| new ParallelGatewaySenderQueueMetaRegion(prQName, ra, null, cache, sender, |
| sender.getStatisticsClock()); |
| |
| try { |
| prQ = (PartitionedRegion) cache.createVMRegion(prQName, ra, |
| new InternalRegionArguments().setInternalMetaRegion(meta).setDestroyLockFlag(true) |
| .setSnapshotInputStream(null).setImageTarget(null)); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("Region created : {} partition Attributes : {}", prQ, |
| prQ.getPartitionAttributes()); |
| } |
| |
| // TODO This should not be set on the PR but on the GatewaySender |
| prQ.enableConflation(sender.isBatchConflationEnabled()); |
| |
| // Before going ahead, make sure all the buckets of shadowPR are |
| // loaded |
| // and primary nodes have been decided. |
| // This is required in case of persistent PR and sender. |
| if (prQ.getLocalMaxMemory() != 0) { |
| Iterator<Integer> itr = prQ.getRegionAdvisor().getBucketSet().iterator(); |
| while (itr.hasNext()) { |
| itr.next(); |
| } |
| } |
| // In case of Replicated Region it may not be necessary. |
| |
| } catch (IOException veryUnLikely) { |
| logger.fatal("Unexpected Exception during init of " + |
| this.getClass(), |
| veryUnLikely); |
| } catch (ClassNotFoundException alsoUnlikely) { |
| logger.fatal("Unexpected Exception during init of " + |
| this.getClass(), |
| alsoUnlikely); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Created queue region: {}", this, prQ); |
| } |
| } else { |
| // in case shadowPR exists already (can be possible when sender is |
| // started from stop operation) |
| if (this.index == 0) // HItesh: for first processor only |
| handleShadowPRExistsScenario(cache, prQ); |
| } |
| /* |
| * Here, enqueueTempEvents need to be invoked when a sender is already running and userPR is |
| * created later. When the flow comes here through start() method of sender i.e. userPR |
| * already exists and sender is started later, the enqueueTempEvents is done in the start() |
| * method of ParallelGatewaySender |
| */ |
| if ((this.index == this.nDispatcher - 1) && this.sender.isRunning()) { |
| ((AbstractGatewaySender) sender).enqueueTempEvents(); |
| } |
| } finally { |
| if (prQ != null) { |
| this.userRegionNameToshadowPRMap.put(userRegion.getFullPath(), prQ); |
| } |
| this.sender.getLifeCycleLock().writeLock().unlock(); |
| } |
| } |
| |
| private static String convertPathToName(String fullPath) { |
| return ""; |
| } |
| |
| public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("{} addShadowPartitionedRegionForUserPR: Attempting to create queue region: {}", |
| this, userPR.getDisplayName()); |
| } |
| this.sender.getLifeCycleLock().writeLock().lock(); |
| |
| PartitionedRegion prQ = null; |
| try { |
| String regionName = userPR.getFullPath(); |
| // Find if there is any parent region for this userPR |
| // if there is then no need to add another q for the same |
| String leaderRegionName = ColocationHelper.getLeaderRegion(userPR).getFullPath(); |
| if (!regionName.equals(leaderRegionName)) { |
| // Fix for defect #50364. Allow user to attach GatewaySender to child PR (without attaching |
| // to leader PR) |
| // though, internally, colocate the GatewaySender's shadowPR with the leader PR in |
| // colocation chain |
| if (!this.userRegionNameToshadowPRMap.containsKey(leaderRegionName)) { |
| addShadowPartitionedRegionForUserPR(ColocationHelper.getLeaderRegion(userPR)); |
| } |
| return; |
| } |
| |
| if (this.userRegionNameToshadowPRMap.containsKey(regionName)) |
| return; |
| |
| if (userPR.getDataPolicy().withPersistence() && !sender.isPersistenceEnabled()) { |
| throw new GatewaySenderException( |
| String.format( |
| "Non persistent gateway sender %s can not be attached to persistent region %s", |
| new Object[] {this.sender.getId(), userPR.getFullPath()})); |
| } |
| |
| InternalCache cache = sender.getCache(); |
| boolean isAccessor = (userPR.getLocalMaxMemory() == 0); |
| |
| final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath()); |
| prQ = (PartitionedRegion) cache.getRegion(prQName); |
| if (prQ == null) { |
| // TODO:REF:Avoid deprecated apis |
| |
| AttributesFactory fact = new AttributesFactory(); |
| fact.setConcurrencyChecksEnabled(false); |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(userPR.getTotalNumberOfBuckets()); |
| pfact.setRedundantCopies(userPR.getRedundantCopies()); |
| pfact.setColocatedWith(regionName); |
| // EITHER set localMaxMemory to 0 for accessor node |
| // OR override shadowPRs default local max memory with the sender's max |
| // queue memory (Fix for bug#44254) |
| int localMaxMemory = isAccessor ? 0 : sender.getMaximumQueueMemory(); |
| pfact.setLocalMaxMemory(localMaxMemory); |
| |
| pfact.setStartupRecoveryDelay(userPR.getPartitionAttributes().getStartupRecoveryDelay()); |
| pfact.setRecoveryDelay(userPR.getPartitionAttributes().getRecoveryDelay()); |
| |
| if (sender.isPersistenceEnabled() && !isAccessor) { |
| fact.setDataPolicy(DataPolicy.PERSISTENT_PARTITION); |
| } |
| |
| fact.setDiskStoreName(sender.getDiskStoreName()); |
| |
| // if persistence is enabled, set the diskSyncronous to whatever user has set |
| // else set it to false |
| if (sender.isPersistenceEnabled()) |
| fact.setDiskSynchronous(sender.isDiskSynchronous()); |
| else { |
| fact.setDiskSynchronous(false); |
| } |
| |
| // allow for no overflow directory |
| EvictionAttributes ea = EvictionAttributes.createLIFOMemoryAttributes( |
| sender.getMaximumQueueMemory(), EvictionAction.OVERFLOW_TO_DISK); |
| |
| fact.setEvictionAttributes(ea); |
| fact.setPartitionAttributes(pfact.create()); |
| |
| final RegionAttributes ra = fact.create(); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Attempting to create queue region: {}", this, prQName); |
| } |
| |
| ParallelGatewaySenderQueueMetaRegion meta = |
| metaRegionFactory.newMetataRegion(cache, prQName, ra, sender); |
| |
| try { |
| prQ = (PartitionedRegion) cache.createVMRegion(prQName, ra, |
| new InternalRegionArguments().setInternalMetaRegion(meta).setDestroyLockFlag(true) |
| .setInternalRegion(true).setSnapshotInputStream(null).setImageTarget(null)); |
| // at this point we should be able to assert prQ == meta; |
| |
| // TODO This should not be set on the PR but on the GatewaySender |
| prQ.enableConflation(sender.isBatchConflationEnabled()); |
| if (isAccessor) |
| return; // return from here if accessor node |
| |
| // Add the overflow statistics to the mbean |
| addOverflowStatisticsToMBean(cache, prQ); |
| |
| // Wait for buckets to be recovered. |
| prQ.shadowPRWaitForBucketRecovery(); |
| |
| } catch (IOException | ClassNotFoundException veryUnLikely) { |
| logger.fatal("Unexpected Exception during init of " + |
| this.getClass(), |
| veryUnLikely); |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Created queue region: {}", this, prQ); |
| } |
| } else { |
| if (isAccessor) |
| return; // return from here if accessor node |
| // in case shadowPR exists already (can be possible when sender is |
| // started from stop operation) |
| if (this.index == 0) // HItesh:for first parallelGatewaySenderQueue only |
| handleShadowPRExistsScenario(cache, prQ); |
| } |
| |
| } finally { |
| if (prQ != null) { |
| this.userRegionNameToshadowPRMap.put(userPR.getFullPath(), prQ); |
| } |
| /* |
| * Here, enqueueTempEvents need to be invoked when a sender is already running and userPR is |
| * created later. When the flow comes here through start() method of sender i.e. userPR |
| * already exists and sender is started later, the enqueueTempEvents is done in the start() |
| * method of ParallelGatewaySender |
| */ |
| if ((this.index == this.nDispatcher - 1) && this.sender.isRunning()) { |
| ((AbstractGatewaySender) sender).enqueueTempEvents(); |
| } |
| afterRegionAdd(userPR); |
| this.sender.getLifeCycleLock().writeLock().unlock(); |
| } |
| } |
| |
| private void addOverflowStatisticsToMBean(Cache cache, PartitionedRegion prQ) { |
| // Get the appropriate mbean and add the eviction and disk region stats to it |
| ManagementService service = ManagementService.getManagementService(cache); |
| if (this.sender.getId().contains(AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX)) { |
| AsyncEventQueueMBean bean = (AsyncEventQueueMBean) service.getLocalAsyncEventQueueMXBean( |
| AsyncEventQueueImpl.getAsyncEventQueueIdFromSenderId(this.sender.getId())); |
| |
| if (bean != null) { |
| // Add the eviction stats |
| bean.getBridge().addOverflowStatistics(prQ.getEvictionStatistics()); |
| |
| // Add the disk region stats |
| bean.getBridge().addOverflowStatistics(prQ.getDiskRegionStats().getStats()); |
| } |
| } else { |
| GatewaySenderMBean bean = |
| (GatewaySenderMBean) service.getLocalGatewaySenderMXBean(this.sender.getId()); |
| |
| if (bean != null) { |
| // Add the eviction stats |
| bean.getBridge().addOverflowStatistics(prQ.getEvictionStatistics()); |
| |
| // Add the disk region stats |
| bean.getBridge().addOverflowStatistics(prQ.getDiskRegionStats().getStats()); |
| } |
| } |
| } |
| |
| /** |
| * This will be case when the sender is started again after stop operation. |
| */ |
| private void handleShadowPRExistsScenario(Cache cache, PartitionedRegion prQ) { |
| // Note: The region will not be null if the sender is started again after stop operation |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: No need to create the region as the region has been retrieved: {}", this, |
| prQ); |
| } |
| // now, clean up the shadowPR's buckets on this node (primary as well as |
| // secondary) for a fresh start |
| Set<BucketRegion> localBucketRegions = prQ.getDataStore().getAllLocalBucketRegions(); |
| for (BucketRegion bucketRegion : localBucketRegions) { |
| bucketRegion.clear(); |
| } |
| } |
| |
| protected void afterRegionAdd(PartitionedRegion userPR) { |
| // nothing |
| } |
| |
| /** |
| * Initialize the thread pool, setting the number of threads that is equal to the number of |
| * processors available to the JVM. |
| */ |
| private void initializeConflationThreadPool() { |
| int poolSize = Runtime.getRuntime().availableProcessors(); |
| conflationExecutor = |
| LoggingExecutors.newFixedThreadPool("WAN Queue Conflation Thread", true, poolSize); |
| } |
| |
| /** |
| * Cleans up the conflation thread pool. Initially, shutdown is done to avoid accepting any newly |
| * submitted tasks. Wait a while for existing tasks to terminate. If the existing tasks still |
| * don't complete, cancel them by calling shutdownNow. |
| */ |
| private void cleanupConflationThreadPool(AbstractGatewaySender sender) { |
| if (conflationExecutor == null) { |
| return; |
| } |
| conflationExecutor.shutdown();// Disable new tasks from being submitted |
| |
| try { |
| if (!conflationExecutor.awaitTermination(1, TimeUnit.SECONDS)) { |
| conflationExecutor.shutdownNow(); // Cancel currently executing tasks |
| // Wait a while for tasks to respond to being cancelled |
| if (!conflationExecutor.awaitTermination(1, TimeUnit.SECONDS)) { |
| logger.warn("Conflation thread pool did not terminate for the GatewaySender : {}", |
| (sender == null ? "all" : sender)); |
| } |
| } |
| } catch (InterruptedException e) { |
| // (Re-)Cancel if current thread also interrupted |
| conflationExecutor.shutdownNow(); |
| // Preserve interrupt status |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| @Override |
| public boolean put(Object object) throws InterruptedException, CacheException { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| boolean putDone = false; |
| // Can this region ever be null? Should we work with regionName and not with region |
| // instance. |
| // It can't be as put is happening on the region and its still under process |
| GatewaySenderEventImpl value = (GatewaySenderEventImpl) object; |
| |
| boolean isDREvent = isDREvent(sender.getCache(), value); |
| |
| Region region = value.getRegion(); |
| String regionPath = null; |
| if (isDREvent) { |
| regionPath = region.getFullPath(); |
| } else { |
| regionPath = ColocationHelper.getLeaderRegion((PartitionedRegion) region).getFullPath(); |
| } |
| if (isDebugEnabled) { |
| logger.debug("Put is for the region {}", region); |
| } |
| if (!this.userRegionNameToshadowPRMap.containsKey(regionPath)) { |
| if (isDebugEnabled) { |
| logger.debug("The userRegionNameToshadowPRMap is {}", userRegionNameToshadowPRMap); |
| } |
| logger.warn( |
| "GatewaySender: Not queuing the event {}, as the region for which this event originated is not yet configured in the GatewaySender", |
| value); |
| // does not put into queue |
| return false; |
| } |
| |
| PartitionedRegion prQ = this.userRegionNameToshadowPRMap.get(regionPath); |
| int bucketId = value.getBucketId(); |
| Object key = null; |
| if (!isDREvent) { |
| key = value.getShadowKey(); |
| |
| if ((Long) key == -1) { |
| // In case of parallel we don't expect |
| // the key to be not set. If it is the case then the event must be coming |
| // through listener, so return. |
| if (isDebugEnabled) { |
| logger.debug("ParallelGatewaySenderOrderedQueue not putting key {} : Value : {}", key, |
| value); |
| } |
| // does not put into queue |
| return false; |
| } |
| } else { |
| key = value.getEventId(); |
| } |
| |
| if (isDebugEnabled) { |
| logger.debug("ParallelGatewaySenderOrderedQueue putting key {} : Value : {}", key, value); |
| } |
| AbstractBucketRegionQueue brq = |
| (AbstractBucketRegionQueue) prQ.getDataStore().getLocalBucketById(bucketId); |
| |
| try { |
| if (brq == null) { |
| // Set the threadInitLevel to BEFORE_INITIAL_IMAGE. |
| final InitializationLevel oldLevel = |
| LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE); |
| try { |
| // Full path of the bucket: |
| |
| final String bucketFullPath = |
| Region.SEPARATOR + PartitionedRegionHelper.PR_ROOT_REGION_NAME + Region.SEPARATOR |
| + prQ.getBucketName(bucketId); |
| |
| brq = (AbstractBucketRegionQueue) prQ.getCache().getRegionByPath(bucketFullPath); |
| if (isDebugEnabled) { |
| logger.debug( |
| "ParallelGatewaySenderOrderedQueue : The bucket in the cache is bucketRegionName : {} bucket : {}", |
| bucketFullPath, brq); |
| } |
| if (brq != null) { |
| brq.getInitializationLock().readLock().lock(); |
| try { |
| putIntoBucketRegionQueue(brq, key, value); |
| putDone = true; |
| } finally { |
| brq.getInitializationLock().readLock().unlock(); |
| } |
| } else if (isDREvent) { |
| // in case of DR with PGS, if shadow bucket is not found event after |
| // above search then it means that bucket is not intended for this |
| // node. So lets not add this event in temp queue event as we are |
| // doing it for PRevent |
| // does not put onto the queue |
| } else { |
| // We have to handle the case where brq is null because the |
| // colocation |
| // chain is getting destroyed one by one starting from child region |
| // i.e this bucket due to moveBucket operation |
| // In that case we don't want to store this event. |
| if (((PartitionedRegion) prQ.getColocatedWithRegion()).getRegionAdvisor() |
| .getBucketAdvisor(bucketId).getShadowBucketDestroyed()) { |
| if (isDebugEnabled) { |
| logger.debug( |
| "ParallelGatewaySenderOrderedQueue not putting key {} : Value : {} as shadowPR bucket is destroyed.", |
| key, value); |
| } |
| // does not put onto the queue |
| } else { |
| /* |
| * This is to prevent data loss, in the scenario when bucket is not available in the |
| * cache but we know that it will be created. |
| */ |
| BlockingQueue tempQueue = null; |
| synchronized (this.bucketToTempQueueMap) { |
| tempQueue = this.bucketToTempQueueMap.get(bucketId); |
| if (tempQueue == null) { |
| tempQueue = new LinkedBlockingQueue(); |
| this.bucketToTempQueueMap.put(bucketId, tempQueue); |
| } |
| } |
| |
| synchronized (tempQueue) { |
| brq = (AbstractBucketRegionQueue) prQ.getCache().getRegionByPath(bucketFullPath); |
| if (brq != null) { |
| brq.getInitializationLock().readLock().lock(); |
| try { |
| putIntoBucketRegionQueue(brq, key, value); |
| putDone = true; |
| } finally { |
| brq.getInitializationLock().readLock().unlock(); |
| } |
| } else { |
| tempQueue.add(value); |
| putDone = true; |
| // For debugging purpose. |
| if (isDebugEnabled) { |
| logger.debug( |
| "The value {} is enqueued to the tempQueue for the BucketRegionQueue.", |
| value); |
| } |
| } |
| } |
| } |
| } |
| |
| } finally { |
| LocalRegion.setThreadInitLevelRequirement(oldLevel); |
| } |
| } else { |
| boolean thisbucketDestroyed = false; |
| |
| if (!isDREvent) { |
| thisbucketDestroyed = |
| ((PartitionedRegion) prQ.getColocatedWithRegion()).getRegionAdvisor() |
| .getBucketAdvisor(bucketId).getShadowBucketDestroyed() || brq.isDestroyed(); |
| } else { |
| thisbucketDestroyed = brq.isDestroyed(); |
| } |
| |
| if (!thisbucketDestroyed) { |
| putIntoBucketRegionQueue(brq, key, value); |
| putDone = true; |
| } else { |
| if (isDebugEnabled) { |
| logger.debug( |
| "ParallelGatewaySenderOrderedQueue not putting key {} : Value : {} as shadowPR bucket is destroyed.", |
| key, value); |
| } |
| // does not put onto the queue |
| } |
| } |
| } finally { |
| notifyEventProcessorIfRequired(); |
| } |
| return putDone; |
| } |
| |
| public void notifyEventProcessorIfRequired() { |
| // putter thread should not take lock every time |
| if (isQueueEmpty) { |
| queueEmptyLock.lock(); |
| try { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Going to notify, isQueueEmpty {}", isQueueEmpty); |
| } |
| if (isQueueEmpty) { |
| isQueueEmpty = false; |
| queueEmptyCondition.signal(); |
| } |
| } finally { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Notified!, isQueueEmpty {}", isQueueEmpty); |
| } |
| queueEmptyLock.unlock(); |
| } |
| } |
| } |
| |
| private void putIntoBucketRegionQueue(AbstractBucketRegionQueue brq, Object key, |
| GatewaySenderEventImpl value) { |
| boolean addedValueToQueue = false; |
| try { |
| if (brq != null) { |
| addedValueToQueue = brq.addToQueue(key, value); |
| // TODO: During merge, ParallelWANstats test failed. On |
| // comment below code test passed. cheetha does not have below code. |
| // need to find out from hcih revision this code came |
| } |
| } catch (BucketNotFoundException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("For bucket {} the current bucket redundancy is {}", brq.getId(), |
| brq.getPartitionedRegion().getRegionAdvisor().getBucketAdvisor(brq.getId()) |
| .getBucketRedundancy()); |
| } |
| } catch (ForceReattemptException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "getInitializedBucketForId: Got ForceReattemptException for {} for bucket = {}", this, |
| brq.getId()); |
| } |
| } finally { |
| if (!addedValueToQueue) { |
| value.release(); |
| } |
| } |
| } |
| |
| /** |
| * This returns queueRegion if there is only one PartitionedRegion using the GatewaySender |
| * Otherwise it returns null. |
| */ |
| @Override |
| public Region getRegion() { |
| return this.userRegionNameToshadowPRMap.size() == 1 |
| ? (Region) this.userRegionNameToshadowPRMap.values().toArray()[0] : null; |
| } |
| |
| public PartitionedRegion getRegion(String fullpath) { |
| return this.userRegionNameToshadowPRMap.get(fullpath); |
| } |
| |
| public PartitionedRegion removeShadowPR(String fullpath) { |
| try { |
| this.sender.getLifeCycleLock().writeLock().lock(); |
| this.sender.setEnqueuedAllTempQueueEvents(false); |
| return this.userRegionNameToshadowPRMap.remove(fullpath); |
| } finally { |
| sender.getLifeCycleLock().writeLock().unlock(); |
| } |
| } |
| |
| public ExecutorService getConflationExecutor() { |
| return this.conflationExecutor; |
| } |
| |
| /** |
| * Returns the set of shadowPR backign this queue. |
| */ |
| public Set<PartitionedRegion> getRegions() { |
| return new HashSet(this.userRegionNameToshadowPRMap.values()); |
| } |
| |
| // TODO: Find optimal way to get Random shadow pr as this will be called in each put and peek. |
| protected PartitionedRegion getRandomShadowPR() { |
| PartitionedRegion prQ = null; |
| if (this.userRegionNameToshadowPRMap.values().size() > 0) { |
| int randomIndex = new Random().nextInt(this.userRegionNameToshadowPRMap.size()); |
| prQ = (PartitionedRegion) this.userRegionNameToshadowPRMap.values().toArray()[randomIndex]; |
| } |
| return prQ; |
| } |
| |
| boolean isDREvent(InternalCache cache, GatewaySenderEventImpl event) { |
| Region region = cache.getRegion(event.getRegionPath()); |
| return region instanceof DistributedRegion; |
| } |
| |
| /** |
| * Take will choose a random BucketRegionQueue which is primary and will take the head element |
| * from it. |
| */ |
| @Override |
| public Object take() throws CacheException, InterruptedException { |
| // merge42180. |
| throw new UnsupportedOperationException(); |
| } |
| |
| protected boolean areLocalBucketQueueRegionsPresent() { |
| boolean bucketsAvailable = false; |
| for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) { |
| if (prQ.getDataStore().getAllLocalBucketRegions().size() > 0) |
| return true; |
| } |
| return false; |
| } |
| |
| private int pickBucketId; |
| |
| protected int getRandomPrimaryBucket(PartitionedRegion prQ) { |
| if (prQ != null) { |
| Set<Map.Entry<Integer, BucketRegion>> allBuckets = prQ.getDataStore().getAllLocalBuckets(); |
| List<Integer> thisProcessorBuckets = new ArrayList<Integer>(); |
| |
| for (Map.Entry<Integer, BucketRegion> bucketEntry : allBuckets) { |
| BucketRegion bucket = bucketEntry.getValue(); |
| if (bucket.getBucketAdvisor().isPrimary()) { |
| int bId = bucket.getId(); |
| if (bId % this.nDispatcher == this.index) { |
| thisProcessorBuckets.add(bId); |
| } |
| } |
| } |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("getRandomPrimaryBucket: total {} for this processor: {}", allBuckets.size(), |
| thisProcessorBuckets.size()); |
| } |
| |
| int nTry = thisProcessorBuckets.size(); |
| |
| while (nTry-- > 0) { |
| if (pickBucketId >= thisProcessorBuckets.size()) |
| pickBucketId = 0; |
| BucketRegionQueue br = |
| getBucketRegionQueueByBucketId(prQ, thisProcessorBuckets.get(pickBucketId++)); |
| if (br != null && br.isReadyForPeek()) { |
| return br.getId(); |
| } |
| } |
| |
| // TODO:REF: instead of shuffle use random number, in this method we are |
| // returning id instead we should return BRQ itself |
| } |
| return -1; |
| } |
| |
| @Override |
| public List take(int batchSize) throws CacheException, InterruptedException { |
| // merge42180 |
| throw new UnsupportedOperationException(); |
| } |
| |
| @Override |
| public void remove() throws CacheException { |
| if (!this.peekedEvents.isEmpty()) { |
| |
| GatewaySenderEventImpl event = this.peekedEvents.remove(); |
| try { |
| PartitionedRegion prQ = null; |
| int bucketId = -1; |
| Object key = null; |
| if (event.getRegion() != null) { |
| if (isDREvent(sender.getCache(), event)) { |
| prQ = this.userRegionNameToshadowPRMap.get(event.getRegion().getFullPath()); |
| bucketId = event.getEventId().getBucketID(); |
| key = event.getEventId(); |
| } else { |
| prQ = this.userRegionNameToshadowPRMap.get(ColocationHelper |
| .getLeaderRegion((PartitionedRegion) event.getRegion()).getFullPath()); |
| bucketId = event.getBucketId(); |
| key = event.getShadowKey(); |
| } |
| } else { |
| String regionPath = event.getRegionPath(); |
| InternalCache cache = this.sender.getCache(); |
| Region region = (PartitionedRegion) cache.getRegion(regionPath); |
| if (region != null && !region.isDestroyed()) { |
| // TODO: We have to get colocated parent region for this region |
| if (region instanceof DistributedRegion) { |
| prQ = this.userRegionNameToshadowPRMap.get(region.getFullPath()); |
| event.getBucketId(); |
| key = event.getEventId(); |
| } else { |
| prQ = this.userRegionNameToshadowPRMap |
| .get(ColocationHelper.getLeaderRegion((PartitionedRegion) region).getFullPath()); |
| event.getBucketId(); |
| key = event.getShadowKey(); |
| } |
| } |
| } |
| |
| if (prQ != null) { |
| destroyEventFromQueue(prQ, bucketId, key); |
| } |
| } finally { |
| try { |
| event.release(); |
| } catch (IllegalStateException e) { |
| logger.error("Exception caught and logged. The thread will continue running", e); |
| } |
| } |
| } |
| } |
| |
| private void destroyEventFromQueue(PartitionedRegion prQ, int bucketId, Object key) { |
| BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId); |
| // TODO : Make sure we dont need to initalize a bucket |
| // before destroying a key from it |
| try { |
| if (brq != null) { |
| brq.destroyKey(key); |
| } |
| stats.decQueueSize(); |
| } catch (EntryNotFoundException e) { |
| if (!this.sender.isBatchConflationEnabled() && logger.isDebugEnabled()) { |
| logger.debug( |
| "ParallelGatewaySenderQueue#remove: Got EntryNotFoundException while removing key {} for {} for bucket = {} for GatewaySender {}", |
| key, this, bucketId, this.sender); |
| } |
| } catch (ForceReattemptException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Bucket :{} moved to other member", bucketId); |
| } |
| } catch (PrimaryBucketException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Primary bucket :{} moved to other member", bucketId); |
| } |
| } catch (RegionDestroyedException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "Caught RegionDestroyedException attempting to remove key {} from bucket {} in {}", key, |
| bucketId, prQ.getFullPath()); |
| } |
| } |
| addRemovedEvent(prQ, bucketId, key); |
| } |
| |
| public void resetLastPeeked() { |
| this.resetLastPeeked = true; |
| |
| // Reset the in progress boolean and queue for peeked events in progress |
| this.peekedEventsProcessingInProgress = false; |
| this.peekedEventsProcessing.clear(); |
| } |
| |
| // Need to improve here.If first peek returns NULL then look in another bucket. |
| @Override |
| public Object peek() throws InterruptedException, CacheException { |
| Object object = null; |
| |
| int bucketId = -1; |
| PartitionedRegion prQ = getRandomShadowPR(); |
| if (prQ != null && prQ.getDataStore().getAllLocalBucketRegions().size() > 0 |
| && ((bucketId = getRandomPrimaryBucket(prQ)) != -1)) { |
| BucketRegionQueue brq; |
| try { |
| brq = ((BucketRegionQueue) prQ.getDataStore().getInitializedBucketForId(null, bucketId)); |
| object = brq.peek(); |
| } catch (BucketRegionQueueUnavailableException e) { |
| return object;// since this is not set, it would be null |
| } catch (ForceReattemptException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Remove: Got ForceReattemptException for {} for bucke = {}", this, bucketId); |
| } |
| } |
| } |
| return object; // OFFHEAP: ok since only callers uses it to check for empty queue |
| } |
| |
| // This method may need synchronization in case it is used by |
| // ConcurrentParallelGatewaySender |
| protected void addRemovedEvent(PartitionedRegion prQ, int bucketId, Object key) { |
| StoppableReentrantLock lock = buckToDispatchLock; |
| if (lock != null) { |
| lock.lock(); |
| boolean wasEmpty = regionToDispatchedKeysMap.isEmpty(); |
| try { |
| Map bucketIdToDispatchedKeys = (Map) regionToDispatchedKeysMap.get(prQ.getFullPath()); |
| if (bucketIdToDispatchedKeys == null) { |
| bucketIdToDispatchedKeys = new ConcurrentHashMap(); |
| regionToDispatchedKeysMap.put(prQ.getFullPath(), bucketIdToDispatchedKeys); |
| } |
| addRemovedEventToMap(bucketIdToDispatchedKeys, bucketId, key); |
| if (wasEmpty) { |
| regionToDispatchedKeysMapEmpty.signal(); |
| } |
| } finally { |
| lock.unlock(); |
| } |
| } |
| } |
| |
| public void sendQueueRemovalMesssageForDroppedEvent(PartitionedRegion prQ, int bucketId, |
| Object key) { |
| final HashMap<String, Map<Integer, List>> temp = new HashMap<String, Map<Integer, List>>(); |
| Map bucketIdToDispatchedKeys = new ConcurrentHashMap(); |
| temp.put(prQ.getFullPath(), bucketIdToDispatchedKeys); |
| addRemovedEventToMap(bucketIdToDispatchedKeys, bucketId, key); |
| Set<InternalDistributedMember> recipients = |
| removalThread.getAllRecipients(sender.getCache(), temp); |
| if (!recipients.isEmpty()) { |
| ParallelQueueRemovalMessage pqrm = new ParallelQueueRemovalMessage(temp); |
| pqrm.setRecipients(recipients); |
| sender.getCache().getInternalDistributedSystem().getDistributionManager().putOutgoing(pqrm); |
| } |
| } |
| |
| private void addRemovedEventToMap(Map bucketIdToDispatchedKeys, int bucketId, Object key) { |
| List dispatchedKeys = (List) bucketIdToDispatchedKeys.get(bucketId); |
| if (dispatchedKeys == null) { |
| dispatchedKeys = new ArrayList<Object>(); |
| bucketIdToDispatchedKeys.put(bucketId, dispatchedKeys); |
| } |
| dispatchedKeys.add(key); |
| } |
| |
| protected void addRemovedEvents(PartitionedRegion prQ, int bucketId, List<Object> shadowKeys) { |
| buckToDispatchLock.lock(); |
| boolean wasEmpty = regionToDispatchedKeysMap.isEmpty(); |
| try { |
| Map bucketIdToDispatchedKeys = (Map) regionToDispatchedKeysMap.get(prQ.getFullPath()); |
| if (bucketIdToDispatchedKeys == null) { |
| bucketIdToDispatchedKeys = new ConcurrentHashMap(); |
| regionToDispatchedKeysMap.put(prQ.getFullPath(), bucketIdToDispatchedKeys); |
| } |
| addRemovedEventsToMap(bucketIdToDispatchedKeys, bucketId, shadowKeys); |
| if (wasEmpty) { |
| regionToDispatchedKeysMapEmpty.signal(); |
| } |
| } finally { |
| buckToDispatchLock.unlock(); |
| } |
| } |
| |
| protected void addRemovedEvents(String prQPath, int bucketId, List<Object> shadowKeys) { |
| buckToDispatchLock.lock(); |
| boolean wasEmpty = regionToDispatchedKeysMap.isEmpty(); |
| try { |
| Map bucketIdToDispatchedKeys = (Map) regionToDispatchedKeysMap.get(prQPath); |
| if (bucketIdToDispatchedKeys == null) { |
| bucketIdToDispatchedKeys = new ConcurrentHashMap(); |
| regionToDispatchedKeysMap.put(prQPath, bucketIdToDispatchedKeys); |
| } |
| addRemovedEventsToMap(bucketIdToDispatchedKeys, bucketId, shadowKeys); |
| if (wasEmpty) { |
| regionToDispatchedKeysMapEmpty.signal(); |
| } |
| } finally { |
| buckToDispatchLock.unlock(); |
| } |
| } |
| |
| private void addRemovedEventsToMap(Map bucketIdToDispatchedKeys, int bucketId, List keys) { |
| List dispatchedKeys = (List) bucketIdToDispatchedKeys.get(bucketId); |
| if (dispatchedKeys == null) { |
| dispatchedKeys = keys == null ? new ArrayList<Object>() : keys; |
| } else { |
| dispatchedKeys.addAll(keys); |
| } |
| bucketIdToDispatchedKeys.put(bucketId, dispatchedKeys); |
| } |
| |
| @Override |
| public List peek(int batchSize) throws InterruptedException, CacheException { |
| throw new UnsupportedOperationException(); |
| } |
| |
| @Override |
| public List peek(int batchSize, int timeToWait) throws InterruptedException, CacheException { |
| final boolean isDebugEnabled = logger.isDebugEnabled(); |
| |
| PartitionedRegion prQ = getRandomShadowPR(); |
| List<GatewaySenderEventImpl> batch = new ArrayList<>(); |
| if (prQ == null || prQ.getLocalMaxMemory() == 0) { |
| try { |
| Thread.sleep(50); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| blockProcessorThreadIfRequired(); |
| return batch; |
| } |
| |
| long start = System.currentTimeMillis(); |
| long end = start + timeToWait; |
| |
| // Add peeked events |
| addPeekedEvents(batch, batchSize); |
| |
| int bId = -1; |
| while (batch.size() < batchSize) { |
| if (areLocalBucketQueueRegionsPresent() && ((bId = getRandomPrimaryBucket(prQ)) != -1)) { |
| GatewaySenderEventImpl object = (GatewaySenderEventImpl) peekAhead(prQ, bId); |
| if (object != null) { |
| GatewaySenderEventImpl copy = object.makeHeapCopyIfOffHeap(); |
| if (copy == null) { |
| if (stats != null) { |
| stats.incEventsNotQueuedConflated(); |
| } |
| continue; |
| } |
| object = copy; |
| } |
| // Conflate here |
| if (object != null) { |
| if (isDebugEnabled) { |
| logger.debug("The gatewayEventImpl in peek is {}", object); |
| } |
| batch.add(object); |
| peekedEvents.add(object); |
| |
| } else { |
| // If time to wait is -1 (don't wait) or time interval has elapsed |
| long currentTime = System.currentTimeMillis(); |
| if (isDebugEnabled) { |
| logger.debug("{}: Peeked object was null. Peek current time: {}", this, currentTime); |
| } |
| if (timeToWait == -1 || (end <= currentTime)) { |
| if (isDebugEnabled) { |
| logger.debug("{}: Peeked object was null.. Peek breaking", this); |
| } |
| break; |
| } |
| if (isDebugEnabled) { |
| logger.debug("{}: Peeked object was null. Peek continuing", this); |
| } |
| continue; |
| } |
| } else { |
| // If time to wait is -1 (don't wait) or time interval has elapsed |
| long currentTime = System.currentTimeMillis(); |
| if (isDebugEnabled) { |
| logger.debug("{}: Peek current time: {}", this, currentTime); |
| } |
| if (timeToWait == -1 || (end <= currentTime)) { |
| if (isDebugEnabled) { |
| logger.debug("{}: Peek breaking", this); |
| } |
| break; |
| } |
| if (isDebugEnabled) { |
| logger.debug("{}: Peek continuing", this); |
| } |
| // Sleep a bit before trying again. |
| try { |
| Thread.sleep(getTimeToSleep(end - currentTime)); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| break; |
| } |
| continue; |
| } |
| } |
| if (isDebugEnabled) { |
| logger.debug("{}: Peeked a batch of {} entries. The size of the queue is {}. localSize is {}", |
| this, batch.size(), size(), localSize()); |
| } |
| if (batch.size() == 0) { |
| blockProcessorThreadIfRequired(); |
| } |
| return batch; |
| } |
| |
| private long getTimeToSleep(long timeToWait) { |
| // Get the minimum of 50 and 5% of the time to wait (which by default is 1000 ms) |
| long timeToSleep = Math.min(50l, ((long) (timeToWait * 0.05))); |
| |
| // If it is 0, then try 50% of the time to wait |
| if (timeToSleep == 0) { |
| timeToSleep = (long) (timeToWait * 0.50); |
| } |
| |
| // If it is still 0, use the time to wait |
| if (timeToSleep == 0) { |
| timeToSleep = timeToWait; |
| } |
| |
| return timeToSleep; |
| } |
| |
| private void addPeekedEvents(List<GatewaySenderEventImpl> batch, int batchSize) { |
| if (this.resetLastPeeked) { |
| |
| // Remove all entries from peekedEvents for buckets that are not longer primary |
| // This will prevent repeatedly trying to dispatch non-primary events |
| for (Iterator<GatewaySenderEventImpl> iterator = peekedEvents.iterator(); iterator |
| .hasNext();) { |
| GatewaySenderEventImpl event = iterator.next(); |
| final int bucketId = event.getBucketId(); |
| final PartitionedRegion region = (PartitionedRegion) event.getRegion(); |
| if (!region.getRegionAdvisor().isPrimaryForBucket(bucketId)) { |
| iterator.remove(); |
| BucketRegionQueue brq = getBucketRegionQueueByBucketId(getRandomShadowPR(), bucketId); |
| if (brq != null) { |
| brq.pushKeyIntoQueue(event.getShadowKey()); |
| } |
| } |
| } |
| |
| if (this.peekedEventsProcessingInProgress) { |
| // Peeked event processing is in progress. This means that the original peekedEvents |
| // contained > batch size events due to a reduction in the batch size. Create a batch |
| // from the peekedEventsProcessing queue. |
| addPreviouslyPeekedEvents(batch, batchSize); |
| } else if (peekedEvents.size() <= batchSize) { |
| // This is the normal case. The connection was lost while processing a batch. |
| // This recreates the batch from the current peekedEvents. |
| batch.addAll(peekedEvents); |
| this.resetLastPeeked = false; |
| } else { |
| // The peekedEvents queue is > batch size. This means that the previous batch size was |
| // reduced due to MessageTooLargeException. Create a batch from the peekedEventsProcessing |
| // queue. |
| this.peekedEventsProcessing.addAll(this.peekedEvents); |
| this.peekedEventsProcessingInProgress = true; |
| addPreviouslyPeekedEvents(batch, batchSize); |
| } |
| if (logger.isDebugEnabled()) { |
| StringBuffer buffer = new StringBuffer(); |
| for (Object ge : batch) { |
| buffer.append("event :"); |
| buffer.append(ge); |
| } |
| logger.debug("Adding already peeked events to the batch {}", buffer); |
| } |
| } |
| } |
| |
| private void addPreviouslyPeekedEvents(List<GatewaySenderEventImpl> batch, int batchSize) { |
| for (int i = 0; i < batchSize; i++) { |
| batch.add(this.peekedEventsProcessing.remove()); |
| if (this.peekedEventsProcessing.isEmpty()) { |
| this.resetLastPeeked = false; |
| this.peekedEventsProcessingInProgress = false; |
| break; |
| } |
| } |
| } |
| |
| protected void blockProcessorThreadIfRequired() throws InterruptedException { |
| queueEmptyLock.lock(); |
| try { |
| if (isQueueEmpty) { // merge44610: this if condition came from cheetah 44610 |
| if (logger.isDebugEnabled()) { |
| logger.debug("Going to wait, till notified."); |
| } |
| // merge44610: this time waiting came from cheetah 44610. In cedar 1000 |
| // is assumed as milliseconds. In cheetah TimeUnitParamter Millisecond |
| // is used. In cheetah stoppable has method to consider timeunit |
| // parameter but cedar does not have such corresponding method |
| queueEmptyCondition.await(1000); |
| // merge44610: this time waiting came from cheetah 44610 |
| } |
| // update the flag so that next time when we come we will block. |
| isQueueEmpty = this.localSizeForProcessor() == 0; |
| } finally { |
| if (logger.isDebugEnabled()) { |
| logger.debug("Going to unblock. isQueueEmpty {}", isQueueEmpty); |
| } |
| queueEmptyLock.unlock(); |
| } |
| } |
| |
| protected Object peekAhead(PartitionedRegion prQ, int bucketId) throws CacheException { |
| Object object = null; |
| BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId); |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Peekahead for the bucket {}", this, bucketId); |
| } |
| try { |
| object = brq.peek(); |
| } catch (BucketRegionQueueUnavailableException e) { |
| // BucketRegionQueue unavailable. Can be due to the BucketRegionQueue being destroyed. |
| return object;// this will be null |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("{}: Peeked object from bucket {} object: {}", this, bucketId, object); |
| } |
| |
| if (object == null) { |
| if (this.stats != null) { |
| this.stats.incEventsNotQueuedConflated(); |
| } |
| } |
| return object; // OFFHEAP: ok since callers are careful to do destroys on region queue after |
| // finished with peeked object. |
| } |
| |
| protected BucketRegionQueue getBucketRegionQueueByBucketId(final PartitionedRegion prQ, |
| final int bucketId) { |
| return (BucketRegionQueue) prQ.getDataStore().getLocalBucketById(bucketId); |
| } |
| |
| public String displayContent() { |
| int size = 0; |
| StringBuffer sb = new StringBuffer(); |
| for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) { |
| if (prQ != null && prQ.getDataStore() != null) { |
| Set<BucketRegion> allLocalBuckets = prQ.getDataStore().getAllLocalBucketRegions(); |
| for (BucketRegion br : allLocalBuckets) { |
| if (br.size() > 0) { |
| sb.append("bucketId=" + br.getId() + ":" + br.keySet() + ";"); |
| } |
| } |
| } |
| } |
| return sb.toString(); |
| } |
| |
| public int localSize() { |
| return localSize(false); |
| } |
| |
| public int localSize(boolean includeSecondary) { |
| int size = 0; |
| for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) { |
| if (prQ != null && prQ.getDataStore() != null) { |
| if (includeSecondary) { |
| size += prQ.getDataStore().getSizeOfLocalBuckets(); |
| } else { |
| size += prQ.getDataStore().getSizeOfLocalPrimaryBuckets(); |
| } |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("The name of the queue region is {} and the size is {}", prQ.getFullPath(), |
| size); |
| } |
| } |
| return size /* + sender.getTmpQueuedEventSize() */; |
| } |
| |
| public int localSizeForProcessor() { |
| int size = 0; |
| for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) { |
| if (((PartitionedRegion) prQ.getRegion()).getDataStore() != null) { |
| Set<BucketRegion> primaryBuckets = |
| ((PartitionedRegion) prQ.getRegion()).getDataStore().getAllLocalPrimaryBucketRegions(); |
| |
| for (BucketRegion br : primaryBuckets) { |
| if (br.getId() % this.nDispatcher == this.index) |
| size += br.size(); |
| } |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("The name of the queue region is {} and the size is {}", prQ.getFullPath(), |
| size); |
| } |
| } |
| return size /* + sender.getTmpQueuedEventSize() */; |
| } |
| |
| @Override |
| public int size() { |
| int size = 0; |
| for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("The name of the queue region is {} and the size is {}. keyset size is {}", |
| prQ.getName(), prQ.size(), prQ.keys().size()); |
| } |
| size += prQ.size(); |
| } |
| |
| return size + sender.getTmpQueuedEventSize(); |
| } |
| |
| @Override |
| public void addCacheListener(CacheListener listener) { |
| for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) { |
| AttributesMutator mutator = prQ.getAttributesMutator(); |
| mutator.addCacheListener(listener); |
| } |
| } |
| |
| @Override |
| public void removeCacheListener() { |
| throw new UnsupportedOperationException(); |
| } |
| |
| @Override |
| public void remove(int batchSize) throws CacheException { |
| for (int i = 0; i < batchSize; i++) { |
| remove(); |
| } |
| } |
| |
| public void conflateEvent(Conflatable conflatableObject, int bucketId, Long tailKey) { |
| ConflationHandler conflationHandler = |
| new ConflationHandler(conflatableObject, bucketId, tailKey); |
| conflationExecutor.execute(conflationHandler); |
| } |
| |
| public long getNumEntriesOverflowOnDiskTestOnly() { |
| long numEntriesOnDisk = 0; |
| for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) { |
| DiskRegionStats diskStats = prQ.getDiskRegionStats(); |
| if (diskStats == null) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: DiskRegionStats for shadow PR is null. Returning the numEntriesOverflowOnDisk as 0", |
| this); |
| } |
| return 0; |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: DiskRegionStats for shadow PR is NOT null. Returning the numEntriesOverflowOnDisk obtained from DiskRegionStats", |
| this); |
| } |
| numEntriesOnDisk += diskStats.getNumOverflowOnDisk(); |
| } |
| return numEntriesOnDisk; |
| } |
| |
| public long getNumEntriesInVMTestOnly() { |
| long numEntriesInVM = 0; |
| for (PartitionedRegion prQ : this.userRegionNameToshadowPRMap.values()) { |
| DiskRegionStats diskStats = prQ.getDiskRegionStats(); |
| if (diskStats == null) { |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: DiskRegionStats for shadow PR is null. Returning the numEntriesInVM as 0", this); |
| } |
| return 0; |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug( |
| "{}: DiskRegionStats for shadow PR is NOT null. Returning the numEntriesInVM obtained from DiskRegionStats", |
| this); |
| } |
| numEntriesInVM += diskStats.getNumEntriesInVM(); |
| } |
| return numEntriesInVM; |
| } |
| |
| /** |
| * This method does the cleanup of any threads, sockets, connection that are held up by the queue. |
| * Note that this cleanup doesn't clean the data held by the queue. |
| */ |
| public void cleanUp() { |
| regionToDispatchedKeysMap.clear(); |
| removalThread.shutdown(); |
| cleanupConflationThreadPool(this.sender); |
| } |
| |
| @Override |
| public void close() { |
| // Because of bug 49060 do not close the regions of a parallel queue |
| } |
| |
| /** |
| * @return the bucketToTempQueueMap |
| */ |
| public Map<Integer, BlockingQueue<GatewaySenderEventImpl>> getBucketToTempQueueMap() { |
| return this.bucketToTempQueueMap; |
| } |
| |
| public static boolean isParallelQueue(String regionName) { |
| return regionName.contains(QSTRING); |
| } |
| |
| public static String getQueueName(String senderId, String regionPath) { |
| return senderId + QSTRING + convertPathToName(regionPath); |
| } |
| |
| public static String getSenderId(String regionName) { |
| int queueStringStart = regionName.indexOf(QSTRING); |
| // The queue id is everything after the leading / and before the QSTRING |
| return regionName.substring(1, queueStringStart); |
| } |
| |
| // TODO:REF: Name for this class should be appropriate? |
| private class BatchRemovalThread extends Thread { |
| /** |
| * boolean to make a shutdown request |
| */ |
| private volatile boolean shutdown = false; |
| |
| private final InternalCache cache; |
| |
| private final ParallelGatewaySenderQueue parallelQueue; |
| |
| /** |
| * Constructor : Creates and initializes the thread |
| */ |
| public BatchRemovalThread(InternalCache c, ParallelGatewaySenderQueue queue) { |
| super("BatchRemovalThread for GatewaySender_" + queue.sender.getId() + "_" + queue.index); |
| this.setDaemon(true); |
| this.cache = c; |
| this.parallelQueue = queue; |
| } |
| |
| private boolean checkCancelled() { |
| if (shutdown) { |
| return true; |
| } |
| if (cache.getCancelCriterion().isCancelInProgress()) { |
| return true; |
| } |
| return false; |
| } |
| |
| @Override |
| public void run() { |
| try { |
| InternalDistributedSystem ids = cache.getInternalDistributedSystem(); |
| DistributionManager dm = ids.getDistributionManager(); |
| for (;;) { |
| try { // be somewhat tolerant of failures |
| if (checkCancelled()) { |
| break; |
| } |
| |
| // TODO : make the thread running time configurable |
| boolean interrupted = Thread.interrupted(); |
| try { |
| synchronized (this) { |
| this.wait(messageSyncInterval); |
| } |
| } catch (InterruptedException e) { |
| interrupted = true; |
| if (checkCancelled()) { |
| break; |
| } |
| break; // desperation...we must be trying to shut |
| // down...? |
| } finally { |
| // Not particularly important since we're exiting |
| // the thread, |
| // but following the pattern is still good |
| // practice... |
| if (interrupted) |
| Thread.currentThread().interrupt(); |
| } |
| |
| if (logger.isDebugEnabled()) { |
| buckToDispatchLock.lock(); |
| try { |
| logger.debug("BatchRemovalThread about to query the batch removal map {}", |
| regionToDispatchedKeysMap); |
| } finally { |
| buckToDispatchLock.unlock(); |
| } |
| } |
| |
| final HashMap<String, Map<Integer, List>> temp; |
| buckToDispatchLock.lock(); |
| try { |
| boolean wasEmpty = regionToDispatchedKeysMap.isEmpty(); |
| while (regionToDispatchedKeysMap.isEmpty()) { |
| regionToDispatchedKeysMapEmpty.await(StoppableCondition.TIME_TO_WAIT); |
| } |
| if (wasEmpty) |
| continue; |
| // TODO: This should be optimized. |
| temp = new HashMap<String, Map<Integer, List>>(regionToDispatchedKeysMap); |
| regionToDispatchedKeysMap.clear(); |
| } finally { |
| buckToDispatchLock.unlock(); |
| } |
| // Get all the data-stores wherever userPRs are present |
| Set<InternalDistributedMember> recipients = getAllRecipients(cache, temp); |
| if (!recipients.isEmpty()) { |
| ParallelQueueRemovalMessage pqrm = new ParallelQueueRemovalMessage(temp); |
| pqrm.setRecipients(recipients); |
| dm.putOutgoing(pqrm); |
| } else { |
| regionToDispatchedKeysMap.putAll(temp); |
| } |
| |
| } // be somewhat tolerant of failures |
| catch (CancelException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("BatchRemovalThread is exiting due to cancellation"); |
| } |
| break; |
| } catch (VirtualMachineError err) { |
| SystemFailure.initiateFailure(err); |
| // If this ever returns, rethrow the error. We're poisoned |
| // now, so don't let this thread continue. |
| throw err; |
| } catch (Throwable t) { |
| Error err; |
| if (t instanceof Error && SystemFailure.isJVMFailureError(err = (Error) t)) { |
| SystemFailure.initiateFailure(err); |
| // If this ever returns, rethrow the error. We're |
| // poisoned now, so don't let this thread continue. |
| throw err; |
| } |
| // Whenever you catch Error or Throwable, you must also |
| // check for fatal JVM error (see above). However, there |
| // is _still_ a possibility that you are dealing with a |
| // cascading error condition, so you also need to check to see if |
| // the JVM is still usable: |
| SystemFailure.checkFailure(); |
| if (checkCancelled()) { |
| break; |
| } |
| if (logger.isDebugEnabled()) { |
| logger.debug("BatchRemovalThread: ignoring exception", t); |
| } |
| } |
| } // for |
| } // ensure exit message is printed |
| catch (CancelException e) { |
| if (logger.isDebugEnabled()) { |
| logger.debug("BatchRemovalThread exiting due to cancellation: " + e); |
| } |
| } finally { |
| logger.info("The QueueRemovalThread is done."); |
| } |
| } |
| |
| private Set<InternalDistributedMember> getAllRecipients(InternalCache cache, Map map) { |
| Set recipients = new ObjectOpenHashSet(); |
| for (Object pr : map.keySet()) { |
| PartitionedRegion partitionedRegion = (PartitionedRegion) cache.getRegion((String) pr); |
| if (partitionedRegion != null && partitionedRegion.getRegionAdvisor() != null) { |
| recipients.addAll(partitionedRegion.getRegionAdvisor().adviseDataStore()); |
| } |
| } |
| return recipients; |
| } |
| |
| /** |
| * shutdown this thread and the caller thread will join this thread |
| */ |
| public void shutdown() { |
| this.shutdown = true; |
| this.interrupt(); |
| boolean interrupted = Thread.interrupted(); |
| try { |
| this.join(15 * 1000); |
| } catch (InterruptedException e) { |
| interrupted = true; |
| } finally { |
| if (interrupted) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| if (this.isAlive()) { |
| logger.warn("QueueRemovalThread ignored cancellation"); |
| } |
| } |
| } |
| |
| protected static class ParallelGatewaySenderQueueMetaRegion extends PartitionedRegion { |
| |
| AbstractGatewaySender sender = null; |
| |
| public ParallelGatewaySenderQueueMetaRegion(String regionName, RegionAttributes attrs, |
| LocalRegion parentRegion, InternalCache cache, AbstractGatewaySender pgSender, |
| StatisticsClock statisticsClock) { |
| super(regionName, attrs, parentRegion, cache, |
| new InternalRegionArguments().setDestroyLockFlag(true).setRecreateFlag(false) |
| .setSnapshotInputStream(null).setImageTarget(null) |
| .setIsUsedForParallelGatewaySenderQueue(true) |
| .setParallelGatewaySender((AbstractGatewaySender) pgSender), |
| statisticsClock); |
| this.sender = (AbstractGatewaySender) pgSender; |
| |
| } |
| |
| @Override |
| public boolean isCopyOnRead() { |
| return false; |
| } |
| |
| // Prevent this region from participating in a TX, bug 38709 |
| @Override |
| public boolean isSecret() { |
| return true; |
| } |
| |
| // Prevent this region from using concurrency checks |
| @Override |
| protected boolean supportsConcurrencyChecks() { |
| return false; |
| } |
| |
| @Override |
| public boolean shouldNotifyBridgeClients() { |
| return false; |
| } |
| |
| @Override |
| public boolean generateEventID() { |
| return false; |
| } |
| |
| @Override |
| protected boolean isUsedForParallelGatewaySenderQueue() { |
| return true; |
| } |
| |
| @Override |
| public AbstractGatewaySender getParallelGatewaySender() { |
| return this.sender; |
| } |
| } |
| |
| public long estimateMemoryFootprint(SingleObjectSizer sizer) { |
| return sizer.sizeof(this) + sizer.sizeof(regionToDispatchedKeysMap) |
| + sizer.sizeof(userRegionNameToshadowPRMap) + sizer.sizeof(bucketToTempQueueMap) |
| + sizer.sizeof(peekedEvents) + sizer.sizeof(conflationExecutor); |
| } |
| |
| public void clear(PartitionedRegion pr, int bucketId) { |
| throw new RuntimeException("This method(clear)is not supported by ParallelGatewaySenderQueue"); |
| } |
| |
| public int size(PartitionedRegion pr, int bucketId) throws ForceReattemptException { |
| throw new RuntimeException("This method(size)is not supported by ParallelGatewaySenderQueue"); |
| } |
| |
| static class MetaRegionFactory { |
| ParallelGatewaySenderQueueMetaRegion newMetataRegion(InternalCache cache, final String prQName, |
| final RegionAttributes ra, AbstractGatewaySender sender) { |
| ParallelGatewaySenderQueueMetaRegion meta = |
| new ParallelGatewaySenderQueueMetaRegion(prQName, ra, null, cache, sender, |
| sender.getStatisticsClock()); |
| return meta; |
| } |
| } |
| } |
