Interim checkin with an ugly example of how threads could be created …

…to handle callback work. The ugly code is commented out.
darshanp · Jan 15, 2013 · ddfbb17 · ddfbb17
1 parent 3162ba2
commit ddfbb17
Show file tree

Hide file tree

Showing 3 changed files with 99 additions and 2 deletions.
diff --git a/src/java/voldemort/store/routed/action/PerformParallelDeleteRequests.java b/src/java/voldemort/store/routed/action/PerformParallelDeleteRequests.java
@@ -197,7 +197,7 @@ private void processResponses(final Map<Integer, Response<ByteArray, Object>> re
                 failureDetector.recordSuccess(response.getNode(), response.getRequestTime());
                 pipelineData.getZoneResponses().add(response.getNode().getZoneId());
 
-                // TODO: Are the next two lines necessary!?!?!?
+                // TODO: Are the next two lines necessary!?!?!? YES, they are.
                 Response<ByteArray, V> rCast = Utils.uncheckedCast(response);
                 pipelineData.getResponses().add(rCast);
 

diff --git a/src/java/voldemort/store/routed/action/PerformParallelPutRequests.java b/src/java/voldemort/store/routed/action/PerformParallelPutRequests.java
@@ -275,6 +275,13 @@ public void requestComplete(Object result, long requestTime) {
                     hintedHandoff.sendHintSerial(node, versionedCopy.getVersion(), slop);
                 }
             }
+            /*-
+            if(isHintedHandoffEnabled() && pipeline.isFinished()) {
+                if(response.getValue() instanceof UnreachableStoreException) {
+                    new Thread(new DoHintedHandoff());
+                }
+            }
+             */
 
             attemptsLatch.countDown();
             blocksLatch.countDown();
@@ -298,6 +305,96 @@ public void requestComplete(Object result, long requestTime) {
                     handleResponseError(response, pipeline, failureDetector);
                 }
             }
+            /*-
+            if(pipeline.isFinished() && response.getValue() instanceof Exception
+               && !(response.getValue() instanceof ObsoleteVersionException)) {
+                new Thread(new DoErrorHandling(response));
+            }
+             */
+
+            /*-
+            if(pipeline.isFinished() && response.getValue() instanceof Exception) {
+                logger.error("OMG DoExceptionHandling.run() invoked!");
+                new Thread(new DoExceptionHandling(response));
+            }
+             */
+        }
+
+        public class DoHintedHandoff implements Runnable {
+
+            DoHintedHandoff() {}
+
+            @Override
+            public void run() {
+                // TODO: remove logger.error...
+                logger.error("OMG DoHintedHandoff.run() invoked!");
+                Slop slop = new Slop(pipelineData.getStoreName(),
+                                     Slop.Operation.PUT,
+                                     key,
+                                     versionedCopy.getValue(),
+                                     transforms,
+                                     node.getId(),
+                                     new Date());
+                pipelineData.addFailedNode(node);
+                hintedHandoff.sendHintSerial(node, versionedCopy.getVersion(), slop);
+            }
+        }
+
+        public class DoErrorHandling implements Runnable {
+
+            Response<ByteArray, Object> response;
+
+            DoErrorHandling(Response<ByteArray, Object> response) {
+                this.response = response;
+            }
+
+            @Override
+            public void run() {
+                // TODO: remove logger.error...
+                logger.error("OMG DoErrorHandling.run() invoked!");
+                if(response.getValue() instanceof InvalidMetadataException) {
+                    pipelineData.reportException((InvalidMetadataException) response.getValue());
+                    logger.warn("Received invalid metadata problem after a successful "
+                                + pipeline.getOperation().getSimpleName() + " call on node "
+                                + node.getId() + ", store '" + pipelineData.getStoreName() + "'");
+                } else {
+                    handleResponseError(response, pipeline, failureDetector);
+                }
+            }
+        }
+
+        public class DoExceptionHandling implements Runnable {
+
+            Response<ByteArray, Object> response;
+
+            DoExceptionHandling(Response<ByteArray, Object> response) {
+                this.response = response;
+            }
+
+            @Override
+            public void run() {
+                // TODO: remove logger.error...
+                logger.error("OMG DoExceptionHandling.run() invoked!");
+                if(response.getValue() instanceof UnreachableStoreException
+                   && isHintedHandoffEnabled()) {
+                    Slop slop = new Slop(pipelineData.getStoreName(),
+                                         Slop.Operation.PUT,
+                                         key,
+                                         versionedCopy.getValue(),
+                                         transforms,
+                                         node.getId(),
+                                         new Date());
+                    pipelineData.addFailedNode(node);
+                    hintedHandoff.sendHintSerial(node, versionedCopy.getVersion(), slop);
+                } else if(response.getValue() instanceof InvalidMetadataException) {
+                    pipelineData.reportException((InvalidMetadataException) response.getValue());
+                    logger.warn("Received invalid metadata problem after a successful "
+                                + pipeline.getOperation().getSimpleName() + " call on node "
+                                + node.getId() + ", store '" + pipelineData.getStoreName() + "'");
+                } else {
+                    handleResponseError(response, pipeline, failureDetector);
+                }
+            }
         }
     }
 }
diff --git a/src/java/voldemort/store/routed/action/PerformParallelRequests.java b/src/java/voldemort/store/routed/action/PerformParallelRequests.java
@@ -136,7 +136,7 @@ else if(pipeline.getOperation() == Operation.GET_VERSIONS)
                 failureDetector.recordSuccess(response.getNode(), response.getRequestTime());
                 pipelineData.getZoneResponses().add(response.getNode().getZoneId());
 
-                // TODO: What about these two operations!? ARe they needed?
+                // TODO: Are the next two lines necessary!?!?!? YES, they are.
                 Response<ByteArray, V> rCast = Utils.uncheckedCast(response);
                 pipelineData.getResponses().add(rCast);
             }