refactored PerformParallelPutAction and related to ensure a slop is r…

…egistered
p6 · May 16, 2013 · 1ba83a1 · 1ba83a1
1 parent cbd4dbd
commit 1ba83a1
Show file tree

Hide file tree

Showing 8 changed files with 252 additions and 178 deletions.
diff --git a/src/java/voldemort/store/routed/PutPipelineData.java b/src/java/voldemort/store/routed/PutPipelineData.java
@@ -16,6 +16,9 @@
 
 package voldemort.store.routed;
 
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
 import voldemort.cluster.Node;
 import voldemort.store.routed.action.PerformSerialPutRequests;
 import voldemort.versioning.Versioned;
@@ -33,6 +36,8 @@ public class PutPipelineData extends BasicPipelineData<Void> {
 
     private long startTimeNs;
 
+    final private List<Node> slopOwnerNodes = new CopyOnWriteArrayList<Node>();
+
     /**
      * Returns the previously determined "master" node. This is the first node
      * in the preference list that succeeded in "putting" the value.
@@ -95,4 +100,23 @@ public long getStartTimeNs() {
         return this.startTimeNs;
     }
 
+    /**
+     * Add a node to the slop owner list for registering slops in hintedHandoff
+     * stage
+     * 
+     * @param node
+     */
+    public void addSlopOwnerNode(Node node) {
+        slopOwnerNodes.add(node);
+    }
+
+    /**
+     * Get list of nodes to register slop for
+     * 
+     * @return list of nodes to register slop for
+     */
+    public List<Node> getSlopOwnerNodes() {
+        return slopOwnerNodes;
+    }
+
 }
diff --git a/src/java/voldemort/store/routed/action/AbstractConfigureNodes.java b/src/java/voldemort/store/routed/action/AbstractConfigureNodes.java
@@ -25,6 +25,7 @@
 import voldemort.store.InsufficientOperationalNodesException;
 import voldemort.store.routed.Pipeline.Event;
 import voldemort.store.routed.PipelineData;
+import voldemort.store.routed.PutPipelineData;
 import voldemort.utils.ByteArray;
 import voldemort.utils.ByteUtils;
 
@@ -61,6 +62,9 @@ protected List<Node> getNodes(ByteArray key) {
             if(failureDetector.isAvailable(node))
                 nodes.add(node);
             else {
+                if(pipelineData instanceof PutPipelineData) {
+                    ((PutPipelineData) pipelineData).addSlopOwnerNode(node);
+                }
                 pipelineData.addFailedNode(node);
                 if(logger.isDebugEnabled()) {
                     logger.debug("Key " + ByteUtils.toHexString(key.get()) + " Node "

diff --git a/src/java/voldemort/store/routed/action/PerformParallelPutRequests.java b/src/java/voldemort/store/routed/action/PerformParallelPutRequests.java
diff --git a/src/java/voldemort/store/routed/action/PerformPutHintedHandoff.java b/src/java/voldemort/store/routed/action/PerformPutHintedHandoff.java
@@ -53,8 +53,8 @@ public PerformPutHintedHandoff(PutPipelineData pipelineData,
     @Override
     public void execute(Pipeline pipeline) {
         Versioned<byte[]> versionedCopy = pipelineData.getVersionedCopy();
-        for(Node failedNode: failedNodes) {
-            int failedNodeId = failedNode.getId();
+        for(Node slopOwnerNode: pipelineData.getSlopOwnerNodes()) {
+            int failedNodeId = slopOwnerNode.getId();
             if(versionedCopy == null) {
                 VectorClock clock = (VectorClock) versioned.getVersion();
                 versionedCopy = new Versioned<byte[]>(versioned.getValue(),
@@ -64,8 +64,9 @@ public void execute(Pipeline pipeline) {
 
             Version version = versionedCopy.getVersion();
             if(logger.isTraceEnabled())
-                logger.trace("Performing hinted handoff for node " + failedNode + ", store "
-                             + pipelineData.getStoreName() + " key " + key + ", version " + version);
+                logger.trace("Performing parallel hinted handoff for node " + slopOwnerNode
+                             + ", store " + pipelineData.getStoreName() + " key " + key
+                             + ", version " + version);
 
             Slop slop = new Slop(pipelineData.getStoreName(),
                                  Slop.Operation.PUT,
@@ -74,7 +75,7 @@ public void execute(Pipeline pipeline) {
                                  transforms,
                                  failedNodeId,
                                  new Date());
-            hintedHandoff.sendHintParallel(failedNode, version, slop);
+            hintedHandoff.sendHintParallel(slopOwnerNode, version, slop);
         }
         pipeline.addEvent(completeEvent);
     }

diff --git a/src/java/voldemort/store/routed/action/PerformSerialPutRequests.java b/src/java/voldemort/store/routed/action/PerformSerialPutRequests.java
@@ -25,6 +25,7 @@
 import voldemort.store.InsufficientOperationalNodesException;
 import voldemort.store.InsufficientZoneResponsesException;
 import voldemort.store.Store;
+import voldemort.store.UnreachableStoreException;
 import voldemort.store.routed.Pipeline;
 import voldemort.store.routed.Pipeline.Event;
 import voldemort.store.routed.PutPipelineData;
@@ -125,6 +126,9 @@ public void execute(Pipeline pipeline) {
                                  + (System.nanoTime() - start) + " ns" + " (keyRef: "
                                  + System.identityHashCode(key) + ")");
 
+                if(e instanceof UnreachableStoreException) {
+                    pipelineData.addSlopOwnerNode(node);
+                }
                 if(handleResponseError(e, node, requestTime, pipeline, failureDetector))
                     return;
             }

diff --git a/src/java/voldemort/store/slop/HintedHandoff.java b/src/java/voldemort/store/slop/HintedHandoff.java
@@ -216,9 +216,13 @@ public boolean sendHintSerial(Node failedNode, Version version, Slop slop) {
                 } catch(UnreachableStoreException e) {
                     failureDetector.recordException(node, (System.nanoTime() - startNs)
                                                           / Time.NS_PER_MS, e);
-                    logger.warn("Error during hinted handoff", e);
+                    logger.warn("Error during hinted handoff. Will try another node", e);
+                } catch(IllegalStateException e) {
+                    logger.warn("Error during hinted handoff. Will try another node", e);
                 } catch(ObsoleteVersionException e) {
                     logger.debug(e, e);
+                } catch(Exception e) {
+                    logger.error("Unknown exception. Will try another node" + e);
                 }
 
                 if(logger.isDebugEnabled())

diff --git a/src/java/voldemort/store/socket/SocketStore.java b/src/java/voldemort/store/socket/SocketStore.java
@@ -284,7 +284,12 @@ private <T> T request(ClientRequest<T> delegate, String operationName) {
             clientRequestExecutor.addClientRequest(blockingClientRequest,
                                                    timeoutMs,
                                                    System.nanoTime() - startTimeNs);
-            blockingClientRequest.await();
+
+            boolean awaitResult = blockingClientRequest.await();
+
+            if(awaitResult == false) {
+                blockingClientRequest.timeOut();
+            }
 
             if(logger.isDebugEnabled())
                 debugMsgStr += "success";

diff --git a/src/java/voldemort/store/socket/clientrequest/BlockingClientRequest.java b/src/java/voldemort/store/socket/clientrequest/BlockingClientRequest.java
@@ -57,8 +57,8 @@ public boolean isComplete() {
         return delegate.isComplete() && latch.getCount() == 0;
     }
 
-    public void await() throws InterruptedException {
-        latch.await(timeoutMs, TimeUnit.MILLISECONDS);
+    public boolean await() throws InterruptedException {
+        return latch.await(timeoutMs, TimeUnit.MILLISECONDS);
     }
 
     public T getResult() throws VoldemortException, IOException {