geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From prhomb...@apache.org
Subject [geode] 01/03: SingleHop implementation for WAN
Date Fri, 16 Nov 2018 23:33:01 GMT
This is an automated email from the ASF dual-hosted git repository.

prhomberg pushed a commit to branch feature/wan_single_hop_wip
in repository https://gitbox.apache.org/repos/asf/geode.git

commit b18c1c59f2b8775df543b5066fa0297bec11d938
Author: Suranjan Kumar <skumar@pivotal.io>
AuthorDate: Fri Oct 30 18:07:23 2015 +0530

        SingleHop implementation for WAN
    
        Keep track of network hop for wan related operations on remote site
        In case of network hop, send the map of GatewayReceiver to primaryBucketIds to sending site
        On sending site, flush all the already peeked events and reditribute the bucket set to processors
        Create extra processors if reuired, bounded by user set totalDispatchers
---
 .../gemfire/cache/client/internal/PoolImpl.java    |   3 +
 .../internal/HDFSParallelGatewaySenderQueue.java   |  16 +--
 .../distributed/internal/DistributionConfig.java   |   2 +-
 .../internal/cache/AbstractBucketRegionQueue.java  |   1 +
 .../gemfire/internal/cache/BucketAdvisor.java      |   3 +-
 .../internal/cache/partitioned/RegionAdvisor.java  |  74 ++++++++++++
 .../sockets/command/GatewayReceiverCommand.java    |  47 +++++++-
 .../internal/cache/wan/AbstractGatewaySender.java  |  11 ++
 .../wan/AbstractGatewaySenderEventProcessor.java   |  79 +++++++++++--
 ...currentParallelGatewaySenderEventProcessor.java |  39 +++----
 .../ConcurrentParallelGatewaySenderQueue.java      |  53 ++++-----
 .../ParallelGatewaySenderEventProcessor.java       |  37 +++++-
 .../wan/parallel/ParallelGatewaySenderQueue.java   | 124 +++++++++++++++++++--
 13 files changed, 408 insertions(+), 81 deletions(-)

diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java
index 8aa5ce2..9c7003f 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/client/internal/PoolImpl.java
@@ -844,6 +844,9 @@ public class PoolImpl implements InternalPool {
     return manager.borrowConnection(45000L);
   }
   
+  public Connection acquireConnectionForWAN(ServerLocation loc) {
+	return manager.borrowConnection(loc,45000L,false);
+  }
   /**
    * Hook to return connections that were acquired using 
    * acquireConnection.
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/HDFSParallelGatewaySenderQueue.java b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/HDFSParallelGatewaySenderQueue.java
index 87e8ce4..8965185 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/HDFSParallelGatewaySenderQueue.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/cache/hdfs/internal/HDFSParallelGatewaySenderQueue.java
@@ -18,11 +18,11 @@
 package com.gemstone.gemfire.cache.hdfs.internal;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 
-
 import com.gemstone.gemfire.cache.CacheException;
 import com.gemstone.gemfire.cache.EntryNotFoundException;
 import com.gemstone.gemfire.cache.Region;
@@ -53,7 +53,7 @@ public class HDFSParallelGatewaySenderQueue extends ParallelGatewaySenderQueue {
   public HDFSParallelGatewaySenderQueue(AbstractGatewaySender sender,
       Set<Region> userPRs, int idx, int nDispatcher) {
      
-    super(sender, userPRs, idx, nDispatcher);
+    super(sender, userPRs, idx, nDispatcher,Collections.<Integer> emptySet());
     //only first dispatcher Hemant?
     if (sender.getBucketSorted() && this.index == 0) {
       rollListTimer = new SystemTimer(sender.getCache().getDistributedSystem(),
@@ -124,7 +124,7 @@ public class HDFSParallelGatewaySenderQueue extends ParallelGatewaySenderQueue {
     if (this.resetLastPeeked) {
       int previousBucketId = -1;
       boolean stillPrimary = true; 
-      Iterator<GatewaySenderEventImpl>  iter = peekedEvents.iterator();
+      Iterator<GatewaySenderEventImpl>  iter = getPeekedEvents().iterator();
       // we need to remove the events of the bucket that are no more primary on 
       // this node as they cannot be persisted from this node. 
       while(iter.hasNext()) {
@@ -213,7 +213,7 @@ public class HDFSParallelGatewaySenderQueue extends ParallelGatewaySenderQueue {
       if (list != null && list.size() != 0 ) {
         for (Object object : list) {
           batch.add(object);
-          peekedEvents.add((HDFSGatewayEventImpl)object);
+          getPeekedEvents().add((HDFSGatewayEventImpl)object);
         }
       }
     }
@@ -252,8 +252,8 @@ public class HDFSParallelGatewaySenderQueue extends ParallelGatewaySenderQueue {
     int destroyed = 0;
     HDFSGatewayEventImpl event = null;
     
-    if (this.peekedEvents.size() > 0)
-      event = (HDFSGatewayEventImpl)this.peekedEvents.remove();
+    if (this.getPeekedEvents().size() > 0)
+      event = (HDFSGatewayEventImpl)this.getPeekedEvents().remove();
     
     while (event != null && destroyed < batchSize) {
       Region currentRegion = event.getRegion();
@@ -269,12 +269,12 @@ public class HDFSParallelGatewaySenderQueue extends ParallelGatewaySenderQueue {
         destroyedSeqNum.add(event.getShadowKey());
         destroyed++;
 
-        if (this.peekedEvents.size() == 0 || (destroyed) >= batchSize) {
+        if (this.getPeekedEvents().size() == 0 || (destroyed) >= batchSize) {
           event = null; 
           break;
         }
 
-        event = (HDFSGatewayEventImpl)this.peekedEvents.remove();
+        event = (HDFSGatewayEventImpl)this.getPeekedEvents().remove();
 
         bucketId = event.getBucketId();
 
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionConfig.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionConfig.java
index 1504421..f21bbdf 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionConfig.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/DistributionConfig.java
@@ -472,7 +472,7 @@ public interface DistributionConfig extends Config, LogConfig {
    * The default log level.
    * <p> Actual value of this constant is {@link InternalLogWriter#CONFIG_LEVEL}.
    */
-  public static final int DEFAULT_LOG_LEVEL = InternalLogWriter.CONFIG_LEVEL;
+  public static final int DEFAULT_LOG_LEVEL = InternalLogWriter.FINE_LEVEL;
   /**
    * The minimum log level.
    * <p> Actual value of this constant is {@link InternalLogWriter#ALL_LEVEL}.
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java
index 2c8f493..487e6be 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/AbstractBucketRegionQueue.java
@@ -527,6 +527,7 @@ public abstract class AbstractBucketRegionQueue extends BucketRegion {
   @Override
   public void afterAcquiringPrimaryState() {
     super.afterAcquiringPrimaryState();
+    // add the serverlocation of primary receiver
     notifyEventProcessor();
   }
   
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketAdvisor.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketAdvisor.java
index 1eaabf1..9fff9da 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketAdvisor.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/BucketAdvisor.java
@@ -1794,7 +1794,8 @@ public final class BucketAdvisor extends CacheDistributionAdvisor  {
     if (!this.pRegion.isShadowPR()) {
       GemFireCacheImpl c = getProxyBucketRegion().getCache();
       List servers = null;
-      servers = c.getCacheServers();
+      servers = c.getCacheServers();// SURANJAN change it to servers and receivrs later on
+      servers = c.getCacheServersAndGatewayReceiver();
 
       HashSet<BucketServerLocation66> serverLocations = new HashSet<BucketServerLocation66>();
       for (Object object : servers) {
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RegionAdvisor.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RegionAdvisor.java
index 03df3fb..0f9ceb6 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RegionAdvisor.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/partitioned/RegionAdvisor.java
@@ -47,6 +47,7 @@ import com.gemstone.gemfire.cache.LowMemoryException;
 import com.gemstone.gemfire.cache.Region;
 import com.gemstone.gemfire.distributed.DistributedMember;
 import com.gemstone.gemfire.distributed.internal.ProfileListener;
+import com.gemstone.gemfire.distributed.internal.ServerLocation;
 import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
 import com.gemstone.gemfire.internal.Assert;
 import com.gemstone.gemfire.internal.InternalDataSerializer;
@@ -303,6 +304,79 @@ public class RegionAdvisor extends CacheDistributionAdvisor
     return bucketToServerLocations;
   }
   
+  /**
+   * get the bucket to primary serverlocation map.
+   * @return
+   */
+  
+  public Map<Integer, BucketServerLocation66> getAllPrimaryClientBucketProfiles() {
+	Map<Integer, BucketServerLocation66> bucketToPrimaryServerLocation = new HashMap<Integer, BucketServerLocation66>();
+    for (Integer bucketId : this.clientBucketProfilesMap.keySet()) {
+      for (BucketProfile profile : this.clientBucketProfilesMap.get(bucketId)) {
+	      ServerBucketProfile cProfile = (ServerBucketProfile)profile;
+	      Set<BucketServerLocation66> bucketServerLocations = cProfile.getBucketServerLocations();
+	      for(BucketServerLocation66 loc: bucketServerLocations) {
+	    	  if(loc.isPrimary()) {
+	    		  bucketToPrimaryServerLocation.put(bucketId, loc);
+	    	  }
+	      }
+      }
+    }
+    
+    if (getPartitionedRegion().isDataStore()) {
+      for (Integer bucketId : getPartitionedRegion().getDataStore().getAllLocalBucketIds()) {
+        BucketProfile profile = getBucketAdvisor(bucketId).getLocalProfile();
+
+        if (logger.isDebugEnabled()) {
+          logger.debug("The local profile is : {}", profile);
+        }
+        
+        if (profile != null) {
+          if (profile instanceof ServerBucketProfile) {
+            ServerBucketProfile cProfile = (ServerBucketProfile)profile;
+            Set<BucketServerLocation66> bucketServerLocations = cProfile
+                .getBucketServerLocations();
+            
+            for(BucketServerLocation66 loc: bucketServerLocations) {
+            	if(loc.isPrimary()) {
+  	    		  bucketToPrimaryServerLocation.put(bucketId, loc);
+  	    	  }
+            }
+          }
+        }
+      }
+    }
+    if (logger.isDebugEnabled()) {
+        logger.debug("The bucket to primary serverlocation is : {}", bucketToPrimaryServerLocation);
+      }
+    return bucketToPrimaryServerLocation;
+  }
+  
+	public Map<ServerLocation, Set<Integer>> getAllPrimaryBucketLocations() {
+		Map<BucketServerLocation66, Set<Integer>> recieverToPrimaryBucketMap = new ConcurrentHashMap<BucketServerLocation66, Set<Integer>>();
+
+		for (BucketServerLocation66 location : getAllPrimaryClientBucketProfiles()
+				.values()) {
+			if (recieverToPrimaryBucketMap.containsKey(location)) {
+				Set<Integer> l = recieverToPrimaryBucketMap.get(location);
+				l.add(location.getBucketId());
+			} else {
+				Set<Integer> bucketList = new HashSet<Integer>();
+				bucketList.add(location.getBucketId());
+				recieverToPrimaryBucketMap.put(location, bucketList);
+			}
+		}
+		
+		Map<ServerLocation, Set<Integer>> recieverToPrimaryBucketMapCopy = new ConcurrentHashMap<ServerLocation, Set<Integer>>();
+		
+		for(BucketServerLocation66 bl: recieverToPrimaryBucketMap.keySet()) {
+			recieverToPrimaryBucketMapCopy.put(
+					new ServerLocation(bl.getHostName(), bl.getPort()),
+					recieverToPrimaryBucketMap.get(bl));
+		}
+		return recieverToPrimaryBucketMapCopy;
+	}
+  
   public ConcurrentHashMap<Integer, Set<ServerBucketProfile>> getAllClientBucketProfilesTest() {
     ConcurrentHashMap<Integer, Set<ServerBucketProfile>> map = new ConcurrentHashMap<Integer, Set<ServerBucketProfile>>();
     Map<Integer, List<BucketServerLocation66>> testMap = this.getAllClientBucketProfiles();
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
index 6c8634d..9334d4f 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/tier/sockets/command/GatewayReceiverCommand.java
@@ -22,6 +22,8 @@ package com.gemstone.gemfire.internal.cache.tier.sockets.command;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import com.gemstone.gemfire.CancelException;
 import com.gemstone.gemfire.cache.EntryNotFoundException;
@@ -32,12 +34,15 @@ import com.gemstone.gemfire.cache.wan.GatewayReceiver;
 import com.gemstone.gemfire.distributed.DistributedSystem;
 import com.gemstone.gemfire.distributed.internal.DistributionStats;
 import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
+import com.gemstone.gemfire.distributed.internal.ServerLocation;
 import com.gemstone.gemfire.i18n.LogWriterI18n;
 import com.gemstone.gemfire.internal.Version;
+import com.gemstone.gemfire.internal.cache.BucketServerLocation66;
 import com.gemstone.gemfire.internal.cache.EntryEventImpl;
 import com.gemstone.gemfire.internal.cache.EventID;
 import com.gemstone.gemfire.internal.cache.KeyWithRegionContext;
 import com.gemstone.gemfire.internal.cache.LocalRegion;
+import com.gemstone.gemfire.internal.cache.PartitionedRegion;
 import com.gemstone.gemfire.internal.cache.tier.CachedRegionHelper;
 import com.gemstone.gemfire.internal.cache.tier.Command;
 import com.gemstone.gemfire.internal.cache.tier.MessageType;
@@ -77,6 +82,8 @@ public class GatewayReceiverCommand extends BaseCommand {
     CachedRegionHelper crHelper = servConn.getCachedRegionHelper();
     GatewayReceiverStats stats = (GatewayReceiverStats)servConn.getCacheServerStats();
     EventID eventId = null;
+    //For single hop assuming that there is only one 
+    // region present
     LocalRegion region = null;
     List<BatchException70> exceptions = new ArrayList<BatchException70>();
     Throwable fatalException = null;
@@ -111,7 +118,7 @@ public class GatewayReceiverCommand extends BaseCommand {
       }
       else {
         logger.warn(LocalizedMessage.create(LocalizedStrings.ProcessBatch_RECEIVED_PROCESS_BATCH_REQUEST_0_THAT_HAS_ALREADY_BEEN_OR_IS_BEING_PROCESSED__THIS_PROCESS_BATCH_REQUEST_IS_BEING_IGNORED, batchId));
-        writeReply(msg, servConn, batchId, numberOfEvents);
+        writeReply(msg, servConn, batchId, numberOfEvents, region);
         return;
       }
       stats.incDuplicateBatchesReceived();
@@ -322,6 +329,10 @@ public class GatewayReceiverCommand extends BaseCommand {
               }
               // Attempt to create the entry
               boolean result = false;
+              
+              if (logger.isDebugEnabled()) {
+                  logger.debug(" SKSKSK Processing event with proxy {}",servConn.getProxyID());
+              }
               result = region.basicBridgeCreate(key, value, isObject, callbackArg,
                       servConn.getProxyID(), false, clientEvent, false); 
               // If the create fails (presumably because it already exists),
@@ -709,7 +720,7 @@ public class GatewayReceiverCommand extends BaseCommand {
       // batch)
       servConn.incrementLatestBatchIdReplied(batchId);
       
-      writeReply(msg, servConn, batchId, numberOfEvents);
+      writeReply(msg, servConn, batchId, numberOfEvents, region);
       servConn.setAsTrue(RESPONDED);
       stats.incWriteProcessBatchResponseTime(DistributionStats.getStatTime()
           - start);
@@ -737,13 +748,37 @@ public class GatewayReceiverCommand extends BaseCommand {
   }
 
   private void writeReply(Message msg, ServerConnection servConn, int batchId,
-      int numberOfEvents) throws IOException {
+      int numberOfEvents, LocalRegion region) throws IOException {
     Message replyMsg = servConn.getResponseMessage();
     replyMsg.setMessageType(MessageType.REPLY);
     replyMsg.setTransactionId(msg.getTransactionId());
-    replyMsg.setNumberOfParts(2);
-    replyMsg.addIntPart(batchId);
-    replyMsg.addIntPart(numberOfEvents);
+    
+    boolean nwhop = false;
+    if(region instanceof PartitionedRegion ) {
+    	PartitionedRegion pr = (PartitionedRegion)region;
+    	nwhop = pr.isNetworkHop().byteValue() != (byte)0;
+    }
+    if (nwhop) {
+    	PartitionedRegion pr = (PartitionedRegion)region;
+    	Map<ServerLocation, Set<Integer>> recieverToPrimaryBucketMap = pr
+    	        .getRegionAdvisor().getAllPrimaryBucketLocations();
+    	if (logger.isDebugEnabled()) {
+    	    logger.debug("Replying with the primary locations as nwhop occurred {}", recieverToPrimaryBucketMap);
+    	}
+        replyMsg.setNumberOfParts(3);
+        replyMsg.addIntPart(batchId);
+        replyMsg.addIntPart(numberOfEvents);
+        replyMsg.addObjPart(recieverToPrimaryBucketMap);
+    }
+    else{
+    	if (logger.isDebugEnabled()) {
+    	    logger.debug("Not Replying with the primary locations as no nwhop occurred");
+    	}
+    	replyMsg.setNumberOfParts(2);
+        replyMsg.addIntPart(batchId);
+        replyMsg.addIntPart(numberOfEvents);
+    }
+    
     replyMsg.setTransactionId(msg.getTransactionId());
     replyMsg.send(servConn);
     servConn.setAsTrue(Command.RESPONDED);
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java
index f1613a3..a10949c 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySender.java
@@ -22,7 +22,9 @@ import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
@@ -57,6 +59,7 @@ import com.gemstone.gemfire.distributed.internal.DistributionAdvisor.Profile;
 import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
 import com.gemstone.gemfire.distributed.internal.ResourceEvent;
 import com.gemstone.gemfire.distributed.internal.ServerLocation;
+import com.gemstone.gemfire.internal.cache.BucketServerLocation66;
 import com.gemstone.gemfire.internal.cache.CachePerfStats;
 import com.gemstone.gemfire.internal.cache.EntryEventImpl;
 import com.gemstone.gemfire.internal.cache.EnumListenerEvent;
@@ -226,6 +229,8 @@ public abstract class AbstractGatewaySender implements GatewaySender,
   
   final Object lockForConcurrentDispatcher = new Object();
 
+  private volatile Map<BucketServerLocation66, List<Integer>> recieverToPrimaryBucketMap = new ConcurrentHashMap<BucketServerLocation66, List<Integer>>();
+
   protected AbstractGatewaySender() {
   }
 
@@ -1309,4 +1314,10 @@ public abstract class AbstractGatewaySender implements GatewaySender,
       this.event.release();
     }
   }
+
+	public void setPrimaryLocations(Map<ServerLocation, Set<Integer>> locations) {
+		if(this.eventProcessor != null) {
+			this.eventProcessor.setPrimaryLocations(locations);
+		}
+	}
 }
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
index b9d877e..e6e9207 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
@@ -24,9 +24,12 @@ import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
 
 import org.apache.logging.log4j.Logger;
 
@@ -43,8 +46,10 @@ import com.gemstone.gemfire.cache.client.internal.pooling.ConnectionDestroyedExc
 import com.gemstone.gemfire.cache.wan.GatewayEventFilter;
 import com.gemstone.gemfire.cache.wan.GatewayQueueEvent;
 import com.gemstone.gemfire.cache.wan.GatewaySender;
+import com.gemstone.gemfire.distributed.internal.ServerLocation;
 import com.gemstone.gemfire.internal.Version;
 import com.gemstone.gemfire.internal.cache.BucketRegion;
+import com.gemstone.gemfire.internal.cache.BucketServerLocation66;
 import com.gemstone.gemfire.internal.cache.Conflatable;
 import com.gemstone.gemfire.internal.cache.DistributedRegion;
 import com.gemstone.gemfire.internal.cache.EntryEventImpl;
@@ -230,11 +235,19 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
   protected void eventQueueRemove() throws CacheException,
       InterruptedException {
     this.queue.remove();
+//    if(this.resumeWhenPeekedEventsEmpty && peekedEvents.isEmpty()) {
+//    	this.resumeWhenPeekedEventsEmpty = false;
+//    }
   }
 
   protected void eventQueueRemove(int size) throws CacheException {
     this.queue.remove(size);
+    
+//    if(this.resumeWhenPeekedEventsEmpty && peekedEvents.isEmpty()) {
+//    	this.resumeWhenPeekedEventsEmpty = false;
+//    }
   }
+  
 
   protected Object eventQueueTake() throws CacheException, InterruptedException {
     throw new UnsupportedOperationException();
@@ -268,7 +281,7 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
     }
     this.isPaused = true;
   }
-
+  
   //merge44957: WHile merging 44957, need this method hence picked up this method from revision 42024.
   public void waitForDispatcherToPause() {
     if (!this.isPaused) {
@@ -320,6 +333,8 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
    */
   private final ConcurrentHashMap<Integer, long[]> failureLogInterval =
       new ConcurrentHashMap<Integer, long[]>();
+
+  private boolean resumeWhenPeekedEventsEmpty = false;
       
   /**
    * The maximum size of {@link #failureLogInterval} beyond which it will start
@@ -403,11 +418,11 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
       if (stopped()) {
         break;
       }
-
+      
       try {
         // Check if paused. If so, wait for resumption
         if (this.isPaused) {
-          waitForResumption();
+        	waitForResumption();
         }
 
         // Peek a batch
@@ -476,7 +491,13 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
               }
             }*/
             }
-            events = this.queue.peek(batchSize, batchTimeInterval);
+            if(this.resumeWhenPeekedEventsEmpty) {
+            	if(this.queue instanceof ParallelGatewaySenderQueue) {
+            		events = ((ParallelGatewaySenderQueue)this.queue).peekAlreadyPeekedEvents();
+            	}
+            }else{
+            	events = this.queue.peek(batchSize, batchTimeInterval);
+            }
           } catch (InterruptedException e) {
             interrupted = true;
             this.sender.getCancelCriterion().checkCancelInProgress(e);
@@ -979,6 +1000,8 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
       eventQueueRemove(events.size());
     }
     
+    //set the latest primary locations to the sender.
+    
   }
   
   public void handleUnSuccessBatchAck(int bId) {
@@ -1241,7 +1264,9 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
     return numEventsDispatched;
   }
   public void clear(PartitionedRegion pr, int bucketId) {
-    ((ParallelGatewaySenderQueue)this.queue).clear(pr, bucketId);
+	  if(this.queue instanceof ParallelGatewaySenderQueue) {
+		  ((ParallelGatewaySenderQueue)this.queue).clear(pr, bucketId);
+	  }
 }
 
 /*public int size(PartitionedRegion pr, int bucketId)
@@ -1250,7 +1275,10 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
 }*/
 
   public void notifyEventProcessorIfRequired(int bucketId) {
-    ((ParallelGatewaySenderQueue) this.queue).notifyEventProcessorIfRequired();
+	  if(this.queue instanceof ParallelGatewaySenderQueue) {
+		  ((ParallelGatewaySenderQueue) this.queue).notifyEventProcessorIfRequired();	  
+	  }
+    
   }
 
   public BlockingQueue<GatewaySenderEventImpl> getBucketTmpQueue(int bucketId) {
@@ -1344,5 +1372,42 @@ public abstract class AbstractGatewaySenderEventProcessor extends Thread {
       }
       return true;
     }
-  }   
+  }
+
+  public void setPrimaryLocations(Map<ServerLocation, Set<Integer>> locations) {
+  }
+
+	public boolean isPeekedEventEmpty() {
+		if (logger.isDebugEnabled()) {
+			logger.debug("SKSKSK Getting called for this processor  " + this);
+		}
+		BlockingQueue peekedEvents = ((ParallelGatewaySenderQueue) this.queue)
+				.getPeekedEvents();
+		if (logger.isDebugEnabled()) {
+			logger.debug("SKSKSK the peeked evets are for this processor  "
+					+ peekedEvents);
+			
+		}
+		
+//		synchronized (peekedEvents) {
+//			while(!peekedEvents.isEmpty()) {
+//				try {
+//					peekedEvents.wait();
+//				} catch (InterruptedException e) {
+//					e.printStackTrace();
+//				}
+//			}
+//		}
+		
+		if (logger.isDebugEnabled()) {
+			logger.debug("SKSKSK WAIT COMPLETE  "
+					+ peekedEvents);
+			
+		}
+		return peekedEvents.isEmpty();
+	}
+
+	public void markResumeWhenPeekedEventEmpty(boolean flag) {
+		this.resumeWhenPeekedEventsEmpty = flag;
+	}   
 }
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderEventProcessor.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderEventProcessor.java
index 970afe3..0cbc741 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderEventProcessor.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderEventProcessor.java
@@ -22,6 +22,9 @@ import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -36,10 +39,6 @@ import com.gemstone.gemfire.InternalGemFireException;
 import com.gemstone.gemfire.cache.CacheException;
 import com.gemstone.gemfire.cache.EntryEvent;
 import com.gemstone.gemfire.cache.Region;
-import com.gemstone.gemfire.cache.hdfs.internal.HDFSBucketRegionQueue;
-import com.gemstone.gemfire.cache.hdfs.internal.HDFSGatewayEventImpl;
-import com.gemstone.gemfire.cache.hdfs.internal.HDFSParallelGatewaySenderQueue;
-import com.gemstone.gemfire.cache.wan.GatewayQueueEvent;
 import com.gemstone.gemfire.internal.cache.EntryEventImpl;
 import com.gemstone.gemfire.internal.cache.EnumListenerEvent;
 import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
@@ -47,7 +46,6 @@ import com.gemstone.gemfire.internal.cache.LocalRegion;
 import com.gemstone.gemfire.internal.cache.RegionQueue;
 import com.gemstone.gemfire.internal.cache.wan.AbstractGatewaySender;
 import com.gemstone.gemfire.internal.cache.wan.AbstractGatewaySenderEventProcessor;
-import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventCallbackDispatcher;
 import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventDispatcher;
 import com.gemstone.gemfire.internal.cache.wan.GatewaySenderException;
 import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
@@ -71,7 +69,8 @@ public class ConcurrentParallelGatewaySenderEventProcessor extends AbstractGatew
 
   protected static final Logger logger = LogService.getLogger();
   
-  protected ParallelGatewaySenderEventProcessor processors[];
+  protected List<ParallelGatewaySenderEventProcessor> processors;
+  
   //private final List<ConcurrentParallelGatewaySenderQueue> concurrentParallelQueues;
   private GemFireException ex = null;
   final int nDispatcher;
@@ -116,13 +115,13 @@ public class ConcurrentParallelGatewaySenderEventProcessor extends AbstractGatew
   }
   
   protected void createProcessors(int dispatcherThreads, Set<Region> targetRs) {
-    processors = new ParallelGatewaySenderEventProcessor[sender.getDispatcherThreads()];
+    processors = new CopyOnWriteArrayList<ParallelGatewaySenderEventProcessor>();
     if (logger.isDebugEnabled()) {
       logger.debug("Creating AsyncEventProcessor");
     }
     for (int i = 0; i < sender.getDispatcherThreads(); i++) {
-      processors[i] = new ParallelGatewaySenderEventProcessor(sender,
-          targetRs, i, sender.getDispatcherThreads());
+      this.processors.add(new ParallelGatewaySenderEventProcessor(sender,
+          targetRs, i, sender.getDispatcherThreads()));
     }
   }
 
@@ -154,7 +153,7 @@ public class ConcurrentParallelGatewaySenderEventProcessor extends AbstractGatew
     	return;
     }
     int pId = bucketId % this.nDispatcher;
-    this.processors[pId].enqueueEvent(operation, event, substituteValue);
+    this.processors.get(pId).enqueueEvent(operation, event, substituteValue);
     
    /* if (getSender().beforeEnque(gatewayQueueEvent)) {
       long start = getSender().getStatistics().startTime();
@@ -178,11 +177,12 @@ public class ConcurrentParallelGatewaySenderEventProcessor extends AbstractGatew
   public void run() {
     final boolean isDebugEnabled = logger.isDebugEnabled();
     
-    for(int i = 0; i < this.processors.length; i++){
+    for(ParallelGatewaySenderEventProcessor p:  this.processors){
+    	int i=0;
       if (isDebugEnabled) {
-        logger.debug("Starting the ParallelProcessors {}", i);
+        logger.debug("Starting the ParallelProcessors {}", i++);
       }
-      this.processors[i].start();
+      p.start();
     }
     try {
       waitForRunningStatus();
@@ -258,8 +258,8 @@ public class ConcurrentParallelGatewaySenderEventProcessor extends AbstractGatew
       stopperCallables.add(new SenderStopperCallable(parallelProcessor));
     }
     
-    ExecutorService stopperService = Executors.newFixedThreadPool(processors.length, threadFactory);
-    try {
+    ExecutorService stopperService = Executors.newFixedThreadPool(processors.size(), threadFactory);
+    try {	
       List<Future<Boolean>> futures = stopperService.invokeAll(stopperCallables);
       for(Future<Boolean> f: futures) {
         try {
@@ -317,6 +317,9 @@ public class ConcurrentParallelGatewaySenderEventProcessor extends AbstractGatew
   public void resumeDispatching() {
     for (ParallelGatewaySenderEventProcessor parallelProcessor : this.processors) {
       parallelProcessor.resumeDispatching();
+      if (logger.isDebugEnabled()) {
+          logger.debug("ParallelGatewaySenderEventProcessor: Resumed dispatching: {}", parallelProcessor);
+        }
     }
     super.resumeDispatching();
     if (logger.isDebugEnabled()) {
@@ -335,9 +338,7 @@ public class ConcurrentParallelGatewaySenderEventProcessor extends AbstractGatew
    */
   public List<ParallelGatewaySenderEventProcessor> getProcessors() {
     List<ParallelGatewaySenderEventProcessor> l = new LinkedList<ParallelGatewaySenderEventProcessor>();
-    for (int i = 0; i < processors.length; i++) {
-      l.add(processors[i]);
-    }
+    l.addAll(processors);
     return l;
   }
 /*
@@ -360,7 +361,7 @@ public class ConcurrentParallelGatewaySenderEventProcessor extends AbstractGatew
  
   @Override
   public GatewaySenderEventDispatcher getDispatcher() {
-    return this.processors[0].getDispatcher();//Suranjan is that fine??
+    return this.processors.get(0).getDispatcher();//Suranjan is that fine??
   }
 
   @Override
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderQueue.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderQueue.java
index 820c40c..0c3ddbb 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderQueue.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ConcurrentParallelGatewaySenderQueue.java
@@ -56,10 +56,10 @@ import com.gemstone.gemfire.internal.size.SingleObjectSizer;
  */
 public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
 
-  private final ParallelGatewaySenderEventProcessor processors[];
+  private final List<ParallelGatewaySenderEventProcessor> processors;
   
   public ConcurrentParallelGatewaySenderQueue(
-		  ParallelGatewaySenderEventProcessor pro[]) {
+		  List<ParallelGatewaySenderEventProcessor> pro) {
     this.processors = pro;
   }
   
@@ -79,15 +79,15 @@ public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
 
   @Override
   public Region getRegion() {
-	  return this.processors[0].getQueue().getRegion();
+	  return this.processors.get(0).getQueue().getRegion();
   }
   
   public PartitionedRegion getRegion(String fullpath) {
-    return processors[0].getRegion(fullpath);
+    return processors.get(0).getRegion(fullpath);
   }
   
   public Set<PartitionedRegion> getRegions() {
-	return ((ParallelGatewaySenderQueue)(processors[0].getQueue())).getRegions();
+	return ((ParallelGatewaySenderQueue)(processors.get(0).getQueue())).getRegions();
   }
 
   @Override
@@ -124,21 +124,21 @@ public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
   @Override
   public int size() {
 	//is that fine??
-	return this.processors[0].getQueue().size();
+	return this.processors.get(0).getQueue().size();
   }
   
   public int localSize() {
-	return ((ParallelGatewaySenderQueue)(processors[0].getQueue())).localSize();
+	return ((ParallelGatewaySenderQueue)(processors.get(0).getQueue())).localSize();
   }
 
   @Override
   public void addCacheListener(CacheListener listener) {
-	  this.processors[0].getQueue().addCacheListener(listener);    
+	  this.processors.get(0).getQueue().addCacheListener(listener);    
   }
 
   @Override
   public void removeCacheListener() {
-    this.processors[0].removeCacheListener();    
+    this.processors.get(0).removeCacheListener();    
   }
 
   @Override
@@ -152,8 +152,8 @@ public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
   
   public long estimateMemoryFootprint(SingleObjectSizer sizer) {
 	long size = 0;
-	for(int i=0; i< processors.length; i++)
-	  size += ((ParallelGatewaySenderQueue)this.processors[i].getQueue()).estimateMemoryFootprint(sizer);
+	for(ParallelGatewaySenderEventProcessor p: this.processors)
+	  size += ((ParallelGatewaySenderQueue)p.getQueue()).estimateMemoryFootprint(sizer);
 	return size;
   }
 
@@ -165,20 +165,20 @@ public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
   }*/
   
   public void removeShadowPR(String prRegionName) {
-  	for(int i =0; i< processors.length; i++){
-   	 processors[i].removeShadowPR(prRegionName);
+	  for(ParallelGatewaySenderEventProcessor p: this.processors){
+		  p.removeShadowPR(prRegionName);
     }
   }
   
   public void addShadowPartitionedRegionForUserPR(PartitionedRegion pr) {
-	for(int i =0; i< processors.length; i++){
-	  processors[i].addShadowPartitionedRegionForUserPR(pr);
+	  for(ParallelGatewaySenderEventProcessor p: this.processors){
+	  p.addShadowPartitionedRegionForUserPR(pr);
 	 }
   }
   
   private ParallelGatewaySenderEventProcessor getPGSProcessor(int bucketId) {
-  	int index = bucketId % this.processors.length;
-  	return processors[index];
+  	int index = bucketId % this.processors.size();
+  	return processors.get(index);
   }
 
   public BlockingQueue<GatewaySenderEventImpl> getBucketTmpQueue(int bucketId) {
@@ -197,11 +197,12 @@ public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
   public void clear(PartitionedRegion pr, int bucketId) {
   	getPGSProcessor(bucketId).clear(pr, bucketId);
   }
-  
-  public void cleanUp() {
-	for(int i=0; i< processors.length; i++)
-	  ((ParallelGatewaySenderQueue)this.processors[i].getQueue()).cleanUp();
-  }
+
+	public void cleanUp() {
+		for (ParallelGatewaySenderEventProcessor p : this.processors) {
+			((ParallelGatewaySenderQueue) p.getQueue()).cleanUp();
+		}
+	}
   
   public void conflateEvent(Conflatable conflatableObject, int bucketId,
       Long tailKey) {
@@ -214,16 +215,16 @@ public class ConcurrentParallelGatewaySenderQueue implements RegionQueue {
   }
   
   public void addShadowPartitionedRegionForUserRR(DistributedRegion userRegion) {
-	for(int i =0; i< processors.length; i++){
-  	 processors[i].addShadowPartitionedRegionForUserRR(userRegion);;
+	  for(ParallelGatewaySenderEventProcessor p: this.processors){
+  	 p.addShadowPartitionedRegionForUserRR(userRegion);;
    }
   }
   
   public long getNumEntriesInVMTestOnly() {
-	return ((ParallelGatewaySenderQueue)(processors[0].getQueue())).getNumEntriesInVMTestOnly();
+	return ((ParallelGatewaySenderQueue)(processors.get(0).getQueue())).getNumEntriesInVMTestOnly();
   }
 	 
   public long getNumEntriesOverflowOnDiskTestOnly() {
-	return ((ParallelGatewaySenderQueue)(processors[0].getQueue())).getNumEntriesOverflowOnDiskTestOnly();
+	return ((ParallelGatewaySenderQueue)(processors.get(0).getQueue())).getNumEntriesOverflowOnDiskTestOnly();
   }
 }
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderEventProcessor.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderEventProcessor.java
index f8780f4..9bc3ae8 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderEventProcessor.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderEventProcessor.java
@@ -20,7 +20,9 @@
 package com.gemstone.gemfire.internal.cache.wan.parallel;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 
 import org.apache.logging.log4j.Logger;
@@ -32,6 +34,8 @@ import com.gemstone.gemfire.cache.hdfs.internal.HDFSBucketRegionQueue;
 import com.gemstone.gemfire.cache.hdfs.internal.HDFSGatewayEventImpl;
 import com.gemstone.gemfire.cache.hdfs.internal.HDFSParallelGatewaySenderQueue;
 import com.gemstone.gemfire.cache.wan.GatewayQueueEvent;
+import com.gemstone.gemfire.distributed.internal.ServerLocation;
+import com.gemstone.gemfire.internal.cache.BucketServerLocation66;
 import com.gemstone.gemfire.internal.cache.Conflatable;
 import com.gemstone.gemfire.internal.cache.DistributedRegion;
 import com.gemstone.gemfire.internal.cache.EntryEventImpl;
@@ -66,6 +70,7 @@ public class ParallelGatewaySenderEventProcessor extends
   
   final int index; 
   final int nDispatcher;
+  Set<Integer> buckets;
 
   protected ParallelGatewaySenderEventProcessor(AbstractGatewaySender sender) {
     super(LoggingThreadGroup.createThreadGroup("Event Processor for GatewaySender_"
@@ -73,10 +78,28 @@ public class ParallelGatewaySenderEventProcessor extends
         "Event Processor for GatewaySender_" + sender.getId(), sender);
     this.index = 0;
     this.nDispatcher = 1;
+    buckets = Collections.emptySet();
     initializeMessageQueue(sender.getId());
     setDaemon(true);
   }
   
+  public synchronized void  setBuckets(Set<Integer> s) {
+	  if (logger.isDebugEnabled()) {
+          logger.debug("ParallelGatewaySenderEventProcessor before setting: {} , {}", s, this.buckets);
+	  }
+	  this.buckets.clear();
+	  this.buckets.addAll(s);
+	  if (logger.isDebugEnabled()) {
+          logger.debug("ParallelGatewaySenderEventProcessor after settingn: {} , {}", s, this.buckets);
+	  }
+	  
+	  ((ParallelGatewaySenderQueue)this.queue).setBuckets(this.buckets);
+	  ((ParallelGatewaySenderQueue)this.queue).setSingleHop();
+  }
+  
+  public Set<Integer> getBuckets() {
+	  return this.buckets;
+  }
   /**
    * use in concurrent scenario where queue is to be shared among all the processors.
    */
@@ -87,6 +110,11 @@ public class ParallelGatewaySenderEventProcessor extends
     this.index = id;
     this.nDispatcher = nDispatcher;
     //this.queue = new ParallelGatewaySenderQueue(sender, userRegions, id, nDispatcher);
+    //SURANJAN for the time being assume 113, it should be totalnumbuckets in the region
+    this.buckets = new HashSet<Integer>();
+    for(int i=index;i< 113; i = i+nDispatcher) {
+      this.buckets.add(i);
+    }
     initializeMessageQueue(sender.getId());
     setDaemon(true);
   }
@@ -107,7 +135,7 @@ public class ParallelGatewaySenderEventProcessor extends
     if (sender.getIsHDFSQueue())
       this.queue = new HDFSParallelGatewaySenderQueue(this.sender, targetRs, this.index, this.nDispatcher);
     else
-      this.queue = new ParallelGatewaySenderQueue(this.sender, targetRs, this.index, this.nDispatcher);
+      this.queue = new ParallelGatewaySenderQueue(this.sender, targetRs, this.index, this.nDispatcher, this.buckets);
     
     if(((ParallelGatewaySenderQueue)queue).localSize() > 0) {
       ((ParallelGatewaySenderQueue)queue).notifyEventProcessorIfRequired();
@@ -238,4 +266,11 @@ public class ParallelGatewaySenderEventProcessor extends
     }
     this.dispatcher = new GatewaySenderEventCallbackDispatcher(this);
   }
+
+  @Override
+  public void setPrimaryLocations(Map<ServerLocation, Set<Integer>> locations) {
+	  this.sender.setPrimaryLocations(locations);
+  }
+  
+
 }
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
index 882a99a..170a965 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
@@ -59,11 +59,13 @@ import com.gemstone.gemfire.cache.asyncqueue.internal.AsyncEventQueueImpl;
 import com.gemstone.gemfire.distributed.internal.DM;
 import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
 import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember;
+import com.gemstone.gemfire.internal.CopyOnWriteHashSet;
 import com.gemstone.gemfire.internal.Version;
 import com.gemstone.gemfire.internal.cache.AbstractBucketRegionQueue;
 import com.gemstone.gemfire.internal.cache.BucketNotFoundException;
 import com.gemstone.gemfire.internal.cache.BucketRegion;
 import com.gemstone.gemfire.internal.cache.BucketRegionQueue;
+import com.gemstone.gemfire.internal.cache.BucketServerLocation66;
 import com.gemstone.gemfire.internal.cache.ColocationHelper;
 import com.gemstone.gemfire.internal.cache.Conflatable;
 import com.gemstone.gemfire.internal.cache.DiskRegionStats;
@@ -142,7 +144,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
   //TODO:REF: name change for thread, as it appears in the log
   private static BatchRemovalThread removalThread = null;
 
-  protected BlockingQueue<GatewaySenderEventImpl> peekedEvents = new LinkedBlockingQueue<GatewaySenderEventImpl>();
+  private BlockingQueue<GatewaySenderEventImpl> peekedEvents = new LinkedBlockingQueue<GatewaySenderEventImpl>();
   
   public final AbstractGatewaySender sender ;
   
@@ -212,7 +214,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
   
   final protected int index; 
   final protected int nDispatcher;
-  
+  protected Set <Integer> buckets;
   /**
    * A transient queue to maintain the eventSeqNum of the events that are to be
    * sent to remote site. It is cleared when the queue is cleared.
@@ -220,9 +222,11 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
   //private final BlockingQueue<Long> eventSeqNumQueue;  
   
   public ParallelGatewaySenderQueue(AbstractGatewaySender sender,
-      Set<Region> userRegions, int idx, int nDispatcher) {
-    this.index = idx;
+      Set<Region> userRegions, int idx, int nDispatcher, Set<Integer> buckets) {
+    
+	this.index = idx;
     this.nDispatcher = nDispatcher;
+    this.buckets = buckets;
     this.stats = sender.getStatistics();
     this.sender = (AbstractGatewaySender)sender;
     
@@ -930,6 +934,9 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
   
   // TODO: Suranjan Find optimal way to get Random shadow pr as this will be called in each put and peek.
   protected PartitionedRegion getRandomShadowPR() {
+	   if (logger.isDebugEnabled()) {
+	          logger.debug("KKKK getRandomPrimaryBucket: {}",userRegionNameToshadowPRMap);
+	   }
     PartitionedRegion prQ = null;
     if (this.userRegionNameToshadowPRMap.values().size() > 0) {
       int randomIndex = new Random().nextInt(this.userRegionNameToshadowPRMap.size());
@@ -992,23 +999,34 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
   }
   
   private int pickBucketId;
-  
+
+  private boolean singleHop;
+
   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(this.buckets.isEmpty() && !singleHop){
+        	if (bId % this.nDispatcher == this.index) {
+        	  thisProcessorBuckets.add(bId);
+        	}
+          }else {
+        	  if(this.buckets.contains(bId)) {
+        		thisProcessorBuckets.add(bId);
+              }
           }
         }
       }
       
       if (logger.isDebugEnabled()) {
+          logger.debug("getRandomPrimaryBucket: allocated {} for this processor", this.buckets);
+        }      
+      
+      if (logger.isDebugEnabled()) {
         logger.debug("getRandomPrimaryBucket: total {} for this processor: {}", allBuckets.size(), thisProcessorBuckets.size());
       }           
       
@@ -1039,6 +1057,19 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
     return -1;
   }
   
+	public Set<Integer> getAllBucket() {
+		PartitionedRegion prQ = getRandomShadowPR();
+		
+		if (prQ != null) {
+			Set<Integer> s = new HashSet<Integer>();
+			for(int i=0;i<prQ.getTotalNumberOfBuckets(); i++) {
+				s.add(i);
+			}
+			return s;
+		}
+		return Collections.EMPTY_SET;
+	}
+  
   @Override
   public List take(int batchSize) throws CacheException, InterruptedException {
     //merge42180
@@ -1097,6 +1128,13 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
         event.release();
       }
     }
+    else {
+			synchronized (peekedEvents) {
+				if (peekedEvents.isEmpty()) {
+					peekedEvents.notifyAll();
+				}
+			}
+		}
   }
 
   private void destroyEventFromQueue(PartitionedRegion prQ, int bucketId,
@@ -1247,6 +1285,36 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
     throw new UnsupportedOperationException();
   }
 
+	public List<GatewaySenderEventImpl> peekAlreadyPeekedEvents() {
+		List batch = new ArrayList();
+		if (this.resetLastPeeked) {
+			batch.addAll(peekedEvents);
+			this.resetLastPeeked = false;
+			final boolean isDebugEnabled = logger.isDebugEnabled();
+
+			if (isDebugEnabled) {
+				StringBuffer buffer = new StringBuffer();
+				for (GatewaySenderEventImpl ge : peekedEvents) {
+					buffer.append("event :");
+					buffer.append(ge);
+				}
+				logger.debug("Adding already peeked events to the batch {}",
+						buffer);
+			}
+		}
+		else {
+			// ideally block
+			try {
+				Thread.sleep(200);
+			} catch (InterruptedException e) {
+				e.printStackTrace();
+			}
+		}
+		
+		
+		return batch;
+	}
+  
   public List peek(int batchSize, int timeToWait) throws InterruptedException,
       CacheException {
     final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -1278,6 +1346,7 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
         logger.debug("Adding already peeked events to the batch {}", buffer);
       }
     }
+    // return batch if we only want peeked events.
     
     int bId = -1;
     while (batch.size() < batchSize) {
@@ -1298,8 +1367,8 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
           }
           batch.add(object);
           peekedEvents.add(object);
-          BucketRegionQueue brq = ((BucketRegionQueue)prQ
-              .getDataStore().getLocalBucketById(bId));
+//          BucketRegionQueue brq = ((BucketRegionQueue)prQ
+//              .getDataStore().getLocalBucketById(bId));
           
           //brq.doLockForPrimary(false);
           
@@ -1429,8 +1498,16 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
         Set<BucketRegion> primaryBuckets = ((PartitionedRegion)prQ.getRegion()).getDataStore().getAllLocalPrimaryBucketRegions();
         
         for (BucketRegion br : primaryBuckets) {
-          if(br.getId() % this.nDispatcher == this.index)
-            size += br.size();
+          int bId = br.getId();
+          if(this.buckets.isEmpty() && !singleHop){
+        	if (bId % this.nDispatcher == this.index) {
+        	  size += br.size();
+        	}
+          }else {//single hop case
+        	if(this.buckets.contains(bId)) {
+        	  size += br.size();
+            }
+          }
         }         
       }
       if (logger.isDebugEnabled()) {
@@ -1473,6 +1550,11 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
     for (int i = 0; i < batchSize; i++) {
       remove();
     }
+	synchronized (peekedEvents) {
+		if (peekedEvents.isEmpty()) {
+			peekedEvents.notifyAll();
+		}
+	}
   }
   
   public void conflateEvent(Conflatable conflatableObject, int bucketId, Long tailKey) {
@@ -1861,4 +1943,22 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
   public int size(PartitionedRegion pr, int bucketId) throws ForceReattemptException{
   	throw new RuntimeException("This method(size)is not supported by ParallelGatewaySenderQueue");
   }
+
+	public void setBuckets(Set<Integer> buckets2) {
+		this.buckets = buckets2;
+		if (logger.isDebugEnabled()) {
+	          logger.debug("ParallelQueue Settign buckets : {} , {}", buckets2, this.buckets);
+	    }
+		
+	}
+
+	public BlockingQueue<GatewaySenderEventImpl> getPeekedEvents() {
+		return peekedEvents;
+	}
+
+	public void setSingleHop() {
+		this.singleHop = true;
+	}
+
+	
 }


Mime
View raw message