HADOOP-9762. Merge change r1506426 from trunk.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1506428 13f79535-47bb-0310-9956-ffa450edef68

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -99,6 +99,9 @@ Release 2.1.0-beta - 2013-07-02
     HADOOP-9763. Extends LightWeightGSet to support eviction of expired 
     elements. (Tsz Wo (Nicholas) SZE via jing9)
 
+    HADOOP-9762. RetryCache utility for implementing RPC retries. 
+    (Suresh Srinivas via jing9)
+
   IMPROVEMENTS
 
     HADOOP-9164. Print paths of loaded native libraries in

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/RetryCache.java

@@ -0,0 +1,293 @@
+/**
+ * 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.hadoop.ipc;
+
+
+import java.util.Arrays;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.util.LightWeightCache;
+import org.apache.hadoop.util.LightWeightGSet;
+import org.apache.hadoop.util.LightWeightGSet.LinkedElement;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Maintains a cache of non-idempotent requests that have been successfully
+ * processed by the RPC server implementation, to handle the retries. A request
+ * is uniquely identified by the unique client ID + call ID of the RPC request.
+ * On receiving retried request, an entry will be found in the
+ * {@link RetryCache} and the previous response is sent back to the request.
+ * <p>
+ * To look an implementation using this cache, see HDFS FSNamesystem class.
+ */
+@InterfaceAudience.Private
+public class RetryCache {
+  public static final Log LOG = LogFactory.getLog(RetryCache.class);
+  /**
+   * CacheEntry is tracked using unique client ID and callId of the RPC request
+   */
+  public static class CacheEntry implements LightWeightCache.Entry {
+    /**
+     * Processing state of the requests
+     */
+    private static byte INPROGRESS = 0;
+    private static byte SUCCESS = 1;
+    private static byte FAILED = 2;
+
+    private volatile byte state = INPROGRESS;
+    
+    // Store uuid as two long for better memory utilization
+    private final long clientIdMsb; // Most signficant bytes
+    private final long clientIdLsb; // Least significant bytes
+    
+    private final int callId;
+    private final long expirationTime;
+    private LightWeightGSet.LinkedElement next;
+
+    CacheEntry(byte[] clientId, int callId, long expirationTime) {
+      Preconditions.checkArgument(clientId.length == 16, "Invalid clientId");
+      // Conver UUID bytes to two longs
+      long tmp = 0;
+      for (int i=0; i<8; i++) {
+        tmp = (tmp << 8) | (clientId[i] & 0xff);
+      }
+      clientIdMsb = tmp;
+      tmp = 0;
+      for (int i=8; i<16; i++) {
+        tmp = (tmp << 8) | (clientId[i] & 0xff);
+      }
+      clientIdLsb = tmp;
+      this.callId = callId;
+      this.expirationTime = expirationTime;
+    }
+
+    private static int hashCode(long value) {
+      return (int)(value ^ (value >>> 32));
+    }
+    
+    @Override
+    public int hashCode() {
+      return (hashCode(clientIdMsb) * 31 + hashCode(clientIdLsb)) * 31 + callId;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj) {
+        return true;
+      }
+      if (!(obj instanceof CacheEntry)) {
+        return false;
+      }
+      CacheEntry other = (CacheEntry) obj;
+      return callId == other.callId && clientIdMsb == other.clientIdMsb
+          && clientIdLsb == other.clientIdLsb;
+    }
+
+    @Override
+    public void setNext(LinkedElement next) {
+      this.next = next;
+    }
+
+    @Override
+    public LinkedElement getNext() {
+      return next;
+    }
+
+    synchronized void completed(boolean success) {
+      state = success ? SUCCESS : FAILED;
+      this.notifyAll();
+    }
+
+    public boolean isSuccess() {
+      return state == SUCCESS;
+    }
+
+    @Override
+    public void setExpirationTime(long timeNano) {
+      // expiration time does not change
+    }
+
+    @Override
+    public long getExpirationTime() {
+      return expirationTime;
+    }
+  }
+
+  /**
+   * CacheEntry with payload that tracks the previous response or parts of
+   * previous response to be used for generating response for retried requests.
+   */
+  public static class CacheEntryWithPayload extends CacheEntry {
+    private Object payload;
+
+    CacheEntryWithPayload(byte[] clientId, int callId, Object payload,
+        long expirationTime) {
+      super(clientId, callId, expirationTime);
+      this.payload = payload;
+    }
+
+    /** Override equals to avoid findbugs warnings */
+    @Override
+    public boolean equals(Object obj) {
+      return super.equals(obj);
+    }
+
+    /** Override hashcode to avoid findbugs warnings */
+    @Override
+    public int hashCode() {
+      return super.hashCode();
+    }
+
+    public Object getPayload() {
+      return payload;
+    }
+  }
+
+  private final LightWeightGSet<CacheEntry, CacheEntry> set;
+  private final long expirationTime;
+
+  /**
+   * Constructor
+   * @param cacheName name to identify the cache by
+   * @param percentage percentage of total java heap space used by this cache
+   * @param expirationTime time for an entry to expire in nanoseconds
+   */
+  public RetryCache(String cacheName, double percentage, long expirationTime) {
+    int capacity = LightWeightGSet.computeCapacity(percentage, cacheName);
+    capacity = capacity > 16 ? capacity : 16;
+    this.set = new LightWeightCache<CacheEntry, CacheEntry>(capacity, capacity,
+        expirationTime, 0);
+    this.expirationTime = expirationTime;
+  }
+
+  private static boolean skipRetryCache() {
+    // Do not track non RPC invocation or RPC requests with
+    // invalid callId or clientId in retry cache
+    return !Server.isRpcInvocation() || Server.getCallId() < 0
+        || Arrays.equals(Server.getClientId(), RpcConstants.DUMMY_CLIENT_ID);
+  }
+
+  /**
+   * This method handles the following conditions:
+   * <ul>
+   * <li>If retry is not to be processed, return null</li>
+   * <li>If there is no cache entry, add a new entry {@code newEntry} and return
+   * it.</li>
+   * <li>If there is an existing entry, wait for its completion. If the
+   * completion state is {@link CacheEntry#FAILED}, the expectation is that the
+   * thread that waited for completion, retries the request. the
+   * {@link CacheEntry} state is set to {@link CacheEntry#INPROGRESS} again.
+   * <li>If the completion state is {@link CacheEntry#SUCCESS}, the entry is
+   * returned so that the thread that waits for it can can return previous
+   * response.</li>
+   * <ul>
+   * 
+   * @return {@link CacheEntry}.
+   */
+  private CacheEntry waitForCompletion(CacheEntry newEntry) {
+    CacheEntry mapEntry = null;
+    synchronized (this) {
+      mapEntry = set.get(newEntry);
+      // If an entry in the cache does not exist, add a new one
+      if (mapEntry == null) {
+        if (LOG.isTraceEnabled()) {
+          LOG.trace("Adding Rpc request clientId "
+              + newEntry.clientIdMsb + newEntry.clientIdLsb + " callId "
+              + newEntry.callId + " to retryCache");
+        }
+        set.put(newEntry);
+        return newEntry;
+      }
+    }
+    // Entry already exists in cache. Wait for completion and return its state
+    Preconditions.checkNotNull(mapEntry,
+        "Entry from the cache should not be null");
+    // Wait for in progress request to complete
+    synchronized (mapEntry) {
+      while (mapEntry.state == CacheEntry.INPROGRESS) {
+        try {
+          mapEntry.wait();
+        } catch (InterruptedException ie) {
+          // Restore the interrupted status
+          Thread.currentThread().interrupt();
+        }
+      }
+      // Previous request has failed, the expectation is is that it will be
+      // retried again.
+      if (mapEntry.state != CacheEntry.SUCCESS) {
+        mapEntry.state = CacheEntry.INPROGRESS;
+      }
+    }
+    return mapEntry;
+  }
+
+  private static CacheEntry newEntry(long expirationTime) {
+    return new CacheEntry(Server.getClientId(), Server.getCallId(),
+        expirationTime);
+  }
+
+  private static CacheEntryWithPayload newEntry(Object payload,
+      long expirationTime) {
+    return new CacheEntryWithPayload(Server.getClientId(), Server.getCallId(),
+        payload, expirationTime);
+  }
+
+  /** Static method that provides null check for retryCache */
+  public static CacheEntry waitForCompletion(RetryCache cache) {
+    if (skipRetryCache()) {
+      return null;
+    }
+    return cache != null ? cache
+        .waitForCompletion(newEntry(cache.expirationTime)) : null;
+  }
+
+  /** Static method that provides null check for retryCache */
+  public static CacheEntryWithPayload waitForCompletion(RetryCache cache,
+      Object payload) {
+    if (skipRetryCache()) {
+      return null;
+    }
+    return (CacheEntryWithPayload) (cache != null ? cache
+        .waitForCompletion(newEntry(payload, cache.expirationTime)) : null);
+  }
+
+  public static void setState(CacheEntry e, boolean success) {
+    if (e == null) {
+      return;
+    }
+    e.completed(success);
+  }
+
+  public static void setState(CacheEntryWithPayload e, boolean success,
+      Object payload) {
+    if (e == null) {
+      return;
+    }
+    e.payload = payload;
+    e.completed(success);
+  }
+
+  public static void clear(RetryCache cache) {
+    if (cache != null) {
+      cache.set.clear();
+    }
+  }
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java

@@ -268,6 +268,12 @@ public abstract class Server {
    */
   private static final ThreadLocal<Call> CurCall = new ThreadLocal<Call>();
   
+  /** Get the current call */
+  @VisibleForTesting
+  public static ThreadLocal<Call> getCurCall() {
+    return CurCall;
+  }
+  
   /**
    * Returns the currently active RPC call's sequential ID number.  A negative
    * call ID indicates an invalid value, such as if there is no currently active
@@ -275,7 +281,7 @@ public abstract class Server {
    * 
    * @return int sequential ID number of currently active RPC call
    */
-  static int getCallId() {
+  public static int getCallId() {
     Call call = CurCall.get();
     return call != null ? call.callId : RpcConstants.INVALID_CALL_ID;
   }
@@ -294,10 +300,8 @@ public abstract class Server {
    */
   public static InetAddress getRemoteIp() {
     Call call = CurCall.get();
-    if (call != null) {
-      return call.connection.getHostInetAddress();
-    }
-    return null;
+    return (call != null && call.connection != null) ? call.connection
+        .getHostInetAddress() : null;
   }
   
   /**
@@ -322,7 +326,8 @@ public abstract class Server {
    */
   public static UserGroupInformation getRemoteUser() {
     Call call = CurCall.get();
-    return (call != null) ? call.connection.user : null;
+    return (call != null && call.connection != null) ? call.connection.user
+        : null;
   }
  
   /** Return true if the invocation was through an RPC.
@@ -460,7 +465,7 @@ public abstract class Server {
   }
 
   /** A call queued for handling. */
-  private static class Call {
+  public static class Call {
     private final int callId;             // the client's call id
     private final int retryCount;        // the retry count of the call
     private final Writable rpcRequest;    // Serialized Rpc request from client
@@ -471,13 +476,13 @@ public abstract class Server {
     private final RPC.RpcKind rpcKind;
     private final byte[] clientId;
 
-    private Call(int id, int retryCount, Writable param, 
+    public Call(int id, int retryCount, Writable param, 
         Connection connection) {
       this(id, retryCount, param, connection, RPC.RpcKind.RPC_BUILTIN,
           RpcConstants.DUMMY_CLIENT_ID);
     }
 
-    private Call(int id, int retryCount, Writable param, Connection connection,
+    public Call(int id, int retryCount, Writable param, Connection connection,
         RPC.RpcKind kind, byte[] clientId) {
       this.callId = id;
       this.retryCount = retryCount;

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestRetryCache.java

@@ -0,0 +1,217 @@
+/**
+ * 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.hadoop.ipc;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.hadoop.ipc.RPC.RpcKind;
+import org.apache.hadoop.ipc.RetryCache.CacheEntryWithPayload;
+import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.util.StringUtils;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Tests for {@link RetryCache}
+ */
+public class TestRetryCache {
+  private static final byte[] CLIENT_ID = StringUtils.getUuidBytes();
+  private static int callId = 100;
+  private static final Random r = new Random();
+  private static final TestServer testServer = new TestServer();
+
+  @Before
+  public void setup() {
+    testServer.resetCounters();
+  }
+
+  static class TestServer {
+    AtomicInteger retryCount = new AtomicInteger();
+    AtomicInteger operationCount = new AtomicInteger();
+    private RetryCache retryCache = new RetryCache("TestRetryCache", 1,
+        100 * 1000 * 1000 * 1000L);
+
+    /**
+     * A server method implemented using {@link RetryCache}.
+     * 
+     * @param input is returned back in echo, if {@code success} is true.
+     * @param failureOuput returned on failure, if {@code success} is false.
+     * @param methodTime time taken by the operation. By passing smaller/larger
+     *          value one can simulate an operation that takes short/long time.
+     * @param success whether this operation completes successfully or not
+     * @return return the input parameter {@code input}, if {@code success} is
+     *         true, else return {@code failureOutput}.
+     */
+    int echo(int input, int failureOutput, long methodTime, boolean success)
+        throws InterruptedException {
+      CacheEntryWithPayload entry = RetryCache.waitForCompletion(retryCache,
+          null);
+      if (entry != null && entry.isSuccess()) {
+        System.out.println("retryCount incremented " + retryCount.get());
+        retryCount.incrementAndGet();
+        return (Integer) entry.getPayload();
+      }
+      try {
+        operationCount.incrementAndGet();
+        if (methodTime > 0) {
+          Thread.sleep(methodTime);
+        }
+      } finally {
+        RetryCache.setState(entry, success, input);
+      }
+      return success ? input : failureOutput;
+    }
+
+    void resetCounters() {
+      retryCount.set(0);
+      operationCount.set(0);
+    }
+  }
+
+  public static Server.Call newCall() {
+    return new Server.Call(++callId, 1, null, null,
+        RpcKind.RPC_PROTOCOL_BUFFER, CLIENT_ID);
+  }
+
+  /**
+   * This simlulates a long server retried operations. Multiple threads start an
+   * operation that takes long time and finally succeeds. The retries in this
+   * case end up waiting for the current operation to complete. All the retries
+   * then complete based on the entry in the retry cache.
+   */
+  @Test
+  public void testLongOperationsSuccessful() throws Exception {
+    // Test long successful operations
+    // There is no entry in cache expected when the first operation starts
+    testOperations(r.nextInt(), 100, 20, true, false, newCall());
+  }
+
+  /**
+   * This simlulates a long server operation. Multiple threads start an
+   * operation that takes long time and finally fails. The retries in this case
+   * end up waiting for the current operation to complete. All the retries end
+   * up performing the operation again.
+   */
+  @Test
+  public void testLongOperationsFailure() throws Exception {
+    // Test long failed operations
+    // There is no entry in cache expected when the first operation starts
+    testOperations(r.nextInt(), 100, 20, false, false, newCall());
+  }
+
+  /**
+   * This simlulates a short server operation. Multiple threads start an
+   * operation that takes very short time and finally succeeds. The retries in
+   * this case do not wait long for the current operation to complete. All the
+   * retries then complete based on the entry in the retry cache.
+   */
+  @Test
+  public void testShortOperationsSuccess() throws Exception {
+    // Test long failed operations
+    // There is no entry in cache expected when the first operation starts
+    testOperations(r.nextInt(), 25, 0, false, false, newCall());
+  }
+
+  /**
+   * This simlulates a short server operation. Multiple threads start an
+   * operation that takes short time and finally fails. The retries in this case
+   * do not wait for the current operation to complete. All the retries end up
+   * performing the operation again.
+   */
+  @Test
+  public void testShortOperationsFailure() throws Exception {
+    // Test long failed operations
+    // There is no entry in cache expected when the first operation starts
+    testOperations(r.nextInt(), 25, 0, false, false, newCall());
+  }
+
+  @Test
+  public void testRetryAfterSuccess() throws Exception {
+    // Previous operation successfully completed
+    Server.Call call = newCall();
+    int input = r.nextInt();
+    Server.getCurCall().set(call);
+    testServer.echo(input, input + 1, 5, true);
+    testOperations(input, 25, 0, true, true, call);
+  }
+
+  @Test
+  public void testRetryAfterFailure() throws Exception {
+    // Previous operation failed
+    Server.Call call = newCall();
+    int input = r.nextInt();
+    Server.getCurCall().set(call);
+    testServer.echo(input, input + 1, 5, false);
+    testOperations(input, 25, 0, false, true, call);
+  }
+
+  public void testOperations(final int input, final int numberOfThreads,
+      final int pause, final boolean success, final boolean attemptedBefore,
+      final Server.Call call) throws InterruptedException, ExecutionException {
+    final int failureOutput = input + 1;
+    ExecutorService executorService = Executors
+        .newFixedThreadPool(numberOfThreads);
+    List<Future<Integer>> list = new ArrayList<Future<Integer>>();
+    for (int i = 0; i < numberOfThreads; i++) {
+      Callable<Integer> worker = new Callable<Integer>() {
+        @Override
+        public Integer call() throws Exception {
+          Server.getCurCall().set(call);
+          Assert.assertEquals(Server.getCurCall().get(), call);
+          int randomPause = pause == 0 ? pause : r.nextInt(pause);
+          return testServer.echo(input, failureOutput, randomPause, success);
+        }
+      };
+      Future<Integer> submit = executorService.submit(worker);
+      list.add(submit);
+    }
+
+    Assert.assertEquals(numberOfThreads, list.size());
+    for (Future<Integer> future : list) {
+      if (success) {
+        Assert.assertEquals(input, future.get().intValue());
+      } else {
+        Assert.assertEquals(failureOutput, future.get().intValue());
+      }
+    }
+
+    if (success) {
+      // If the operation was successful, all the subsequent operations
+      // by other threads should be retries. Operation count should be 1.
+      int retries = numberOfThreads + (attemptedBefore ? 0 : -1);
+      Assert.assertEquals(1, testServer.operationCount.get());
+      Assert.assertEquals(retries, testServer.retryCount.get());
+    } else {
+      // If the operation failed, all the subsequent operations
+      // should execute once more, hence the retry count should be 0 and
+      // operation count should be the number of tries
+      int opCount = numberOfThreads + (attemptedBefore ? 1 : 0);
+      Assert.assertEquals(opCount, testServer.operationCount.get());
+      Assert.assertEquals(0, testServer.retryCount.get());
+    }
+  }
+}