You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by ar...@apache.org on 2014/02/21 21:53:35 UTC
svn commit: r1570703 - in
/hadoop/common/trunk/hadoop-common-project/hadoop-common: ./
src/main/java/org/apache/hadoop/fs/ src/main/java/org/apache/hadoop/ipc/
src/test/java/org/apache/hadoop/ipc/
Author: arp
Date: Fri Feb 21 20:53:34 2014
New Revision: 1570703
URL: http://svn.apache.org/r1570703
Log:
HADOOP-10278. Refactor to make CallQueue pluggable. (Contributed by Chris Li)
Added:
hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallQueueManager.java
hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestCallQueueManager.java
Modified:
hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt
hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java
hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt?rev=1570703&r1=1570702&r2=1570703&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/CHANGES.txt Fri Feb 21 20:53:34 2014
@@ -345,6 +345,9 @@ Release 2.5.0 - UNRELEASED
IMPROVEMENTS
+ HADOOP-10278. Refactor to make CallQueue pluggable. (Chris Li via
+ Arpit Agarwal)
+
OPTIMIZATIONS
BUG FIXES
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java?rev=1570703&r1=1570702&r2=1570703&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java Fri Feb 21 20:53:34 2014
@@ -82,6 +82,14 @@ public class CommonConfigurationKeys ext
/** Default value for IPC_SERVER_HANDLER_QUEUE_SIZE_KEY */
public static final int IPC_SERVER_HANDLER_QUEUE_SIZE_DEFAULT = 100;
+ /**
+ * CallQueue related settings. These are not used directly, but rather
+ * combined with a namespace and port. For instance:
+ * IPC_CALLQUEUE_NAMESPACE + ".8020." + IPC_CALLQUEUE_IMPL_KEY
+ */
+ public static final String IPC_CALLQUEUE_NAMESPACE = "ipc";
+ public static final String IPC_CALLQUEUE_IMPL_KEY = "callqueue.impl";
+
/** Internal buffer size for Lzo compressor/decompressors */
public static final String IO_COMPRESSION_CODEC_LZO_BUFFERSIZE_KEY =
"io.compression.codec.lzo.buffersize";
Added: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallQueueManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallQueueManager.java?rev=1570703&view=auto
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallQueueManager.java (added)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallQueueManager.java Fri Feb 21 20:53:34 2014
@@ -0,0 +1,159 @@
+/**
+ * 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.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.TimeUnit;
+
+import java.lang.reflect.Constructor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+/**
+ * Abstracts queue operations for different blocking queues.
+ */
+public class CallQueueManager<E> {
+ public static final Log LOG = LogFactory.getLog(CallQueueManager.class);
+
+ // Atomic refs point to active callQueue
+ // We have two so we can better control swapping
+ private final AtomicReference<BlockingQueue<E>> putRef;
+ private final AtomicReference<BlockingQueue<E>> takeRef;
+
+ public CallQueueManager(Class backingClass, int maxQueueSize,
+ String namespace, Configuration conf) {
+ BlockingQueue<E> bq = createCallQueueInstance(backingClass,
+ maxQueueSize, namespace, conf);
+ this.putRef = new AtomicReference<BlockingQueue<E>>(bq);
+ this.takeRef = new AtomicReference<BlockingQueue<E>>(bq);
+ LOG.info("Using callQueue " + backingClass);
+ }
+
+ @SuppressWarnings("unchecked")
+ private BlockingQueue<E> createCallQueueInstance(Class theClass, int maxLen,
+ String ns, Configuration conf) {
+
+ // Used for custom, configurable callqueues
+ try {
+ Constructor ctor = theClass.getDeclaredConstructor(int.class, String.class,
+ Configuration.class);
+ return (BlockingQueue<E>)ctor.newInstance(maxLen, ns, conf);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ }
+
+ // Used for LinkedBlockingQueue, ArrayBlockingQueue, etc
+ try {
+ Constructor ctor = theClass.getDeclaredConstructor(int.class);
+ return (BlockingQueue<E>)ctor.newInstance(maxLen);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ }
+
+ // Last attempt
+ try {
+ Constructor ctor = theClass.getDeclaredConstructor();
+ return (BlockingQueue<E>)ctor.newInstance();
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ }
+
+ // Nothing worked
+ throw new RuntimeException(theClass.getName() +
+ " could not be constructed.");
+ }
+
+ /**
+ * Insert e into the backing queue or block until we can.
+ * If we block and the queue changes on us, we will insert while the
+ * queue is drained.
+ */
+ public void put(E e) throws InterruptedException {
+ putRef.get().put(e);
+ }
+
+ /**
+ * Retrieve an E from the backing queue or block until we can.
+ * Guaranteed to return an element from the current queue.
+ */
+ public E take() throws InterruptedException {
+ E e = null;
+
+ while (e == null) {
+ e = takeRef.get().poll(1000L, TimeUnit.MILLISECONDS);
+ }
+
+ return e;
+ }
+
+ public int size() {
+ return takeRef.get().size();
+ }
+
+ /**
+ * Replaces active queue with the newly requested one and transfers
+ * all calls to the newQ before returning.
+ */
+ public synchronized void swapQueue(Class queueClassToUse, int maxSize,
+ String ns, Configuration conf) {
+ BlockingQueue<E> newQ = createCallQueueInstance(queueClassToUse, maxSize,
+ ns, conf);
+
+ // Our current queue becomes the old queue
+ BlockingQueue<E> oldQ = putRef.get();
+
+ // Swap putRef first: allow blocked puts() to be unblocked
+ putRef.set(newQ);
+
+ // Wait for handlers to drain the oldQ
+ while (!queueIsReallyEmpty(oldQ)) {}
+
+ // Swap takeRef to handle new calls
+ takeRef.set(newQ);
+
+ LOG.info("Old Queue: " + stringRepr(oldQ) + ", " +
+ "Replacement: " + stringRepr(newQ));
+ }
+
+ /**
+ * Checks if queue is empty by checking at two points in time.
+ * This doesn't mean the queue might not fill up at some point later, but
+ * it should decrease the probability that we lose a call this way.
+ */
+ private boolean queueIsReallyEmpty(BlockingQueue q) {
+ boolean wasEmpty = q.isEmpty();
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException ie) {
+ return false;
+ }
+ return q.isEmpty() && wasEmpty;
+ }
+
+ private String stringRepr(Object o) {
+ return o.getClass().getName() + '@' + Integer.toHexString(o.hashCode());
+ }
+}
Modified: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java?rev=1570703&r1=1570702&r2=1570703&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java (original)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/Server.java Fri Feb 21 20:53:34 2014
@@ -365,7 +365,7 @@ public abstract class Server {
private final boolean tcpNoDelay; // if T then disable Nagle's Algorithm
volatile private boolean running = true; // true while server runs
- private BlockingQueue<Call> callQueue; // queued calls
+ private CallQueueManager<Call> callQueue;
// maintains the set of client connections and handles idle timeouts
private ConnectionManager connectionManager;
@@ -469,6 +469,19 @@ public abstract class Server {
return serviceAuthorizationManager;
}
+ /*
+ * Refresh the call queue
+ */
+ public synchronized void refreshCallQueue(Configuration conf) {
+ // Create the next queue
+ String prefix = CommonConfigurationKeys.IPC_CALLQUEUE_NAMESPACE + "." +
+ this.port;
+ Class queueClassToUse = conf.getClass(prefix + "." +
+ CommonConfigurationKeys.IPC_CALLQUEUE_IMPL_KEY, LinkedBlockingQueue.class);
+
+ callQueue.swapQueue(queueClassToUse, maxQueueSize, prefix, conf);
+ }
+
/** A call queued for handling. */
public static class Call {
private final int callId; // the client's call id
@@ -2193,7 +2206,15 @@ public abstract class Server {
this.readerPendingConnectionQueue = conf.getInt(
CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_KEY,
CommonConfigurationKeys.IPC_SERVER_RPC_READ_CONNECTION_QUEUE_SIZE_DEFAULT);
- this.callQueue = new LinkedBlockingQueue<Call>(maxQueueSize);
+
+ // Setup appropriate callqueue
+ String prefix = CommonConfigurationKeys.IPC_CALLQUEUE_NAMESPACE + "." +
+ this.port;
+ Class queueClassToUse = conf.getClass(prefix + "." +
+ CommonConfigurationKeys.IPC_CALLQUEUE_IMPL_KEY, LinkedBlockingQueue.class);
+ this.callQueue = new CallQueueManager<Call>(queueClassToUse, maxQueueSize,
+ prefix, conf);
+
this.secretManager = (SecretManager<TokenIdentifier>) secretManager;
this.authorize =
conf.getBoolean(CommonConfigurationKeys.HADOOP_SECURITY_AUTHORIZATION,
Added: hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestCallQueueManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestCallQueueManager.java?rev=1570703&view=auto
==============================================================================
--- hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestCallQueueManager.java (added)
+++ hadoop/common/trunk/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/ipc/TestCallQueueManager.java Fri Feb 21 20:53:34 2014
@@ -0,0 +1,223 @@
+/**
+ * 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.concurrent.LinkedBlockingQueue;
+import java.util.HashMap;
+import java.util.ArrayList;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+import org.junit.Before;
+import org.junit.After;
+
+public class TestCallQueueManager {
+ private CallQueueManager<FakeCall> manager;
+
+ public class FakeCall {
+ public final int tag; // Can be used for unique identification
+
+ public FakeCall(int tag) {
+ this.tag = tag;
+ }
+ }
+
+ /**
+ * Putter produces FakeCalls
+ */
+ public class Putter implements Runnable {
+ private final CallQueueManager<FakeCall> cq;
+
+ public final int tag;
+ public volatile int callsAdded = 0; // How many calls we added, accurate unless interrupted
+ private final int maxCalls;
+
+ private boolean isRunning = true;
+
+ public Putter(CallQueueManager<FakeCall> aCq, int maxCalls, int tag) {
+ this.maxCalls = maxCalls;
+ this.cq = aCq;
+ this.tag = tag;
+ }
+
+ public void run() {
+ try {
+ // Fill up to max (which is infinite if maxCalls < 0)
+ while (isRunning && (callsAdded < maxCalls || maxCalls < 0)) {
+ cq.put(new FakeCall(this.tag));
+ callsAdded++;
+ }
+ } catch (InterruptedException e) {
+ return;
+ }
+ }
+
+ public void stop() {
+ this.isRunning = false;
+ }
+ }
+
+ /**
+ * Taker consumes FakeCalls
+ */
+ public class Taker implements Runnable {
+ private final CallQueueManager<FakeCall> cq;
+
+ public final int tag; // if >= 0 means we will only take the matching tag, and put back
+ // anything else
+ public volatile int callsTaken = 0; // total calls taken, accurate if we aren't interrupted
+ public volatile FakeCall lastResult = null; // the last thing we took
+ private final int maxCalls; // maximum calls to take
+
+ public Taker(CallQueueManager<FakeCall> aCq, int maxCalls, int tag) {
+ this.maxCalls = maxCalls;
+ this.cq = aCq;
+ this.tag = tag;
+ }
+
+ public void run() {
+ try {
+ // Take while we don't exceed maxCalls, or if maxCalls is undefined (< 0)
+ while (callsTaken < maxCalls || maxCalls < 0) {
+ FakeCall res = cq.take();
+
+ if (tag >= 0 && res.tag != this.tag) {
+ // This call does not match our tag, we should put it back and try again
+ cq.put(res);
+ } else {
+ callsTaken++;
+ lastResult = res;
+ }
+ }
+ } catch (InterruptedException e) {
+ return;
+ }
+ }
+ }
+
+ // Assert we can take exactly the numberOfTakes
+ public void assertCanTake(CallQueueManager<FakeCall> cq, int numberOfTakes,
+ int takeAttempts) throws InterruptedException {
+
+ Taker taker = new Taker(cq, takeAttempts, -1);
+ Thread t = new Thread(taker);
+ t.start();
+ t.join(100);
+
+ assertEquals(taker.callsTaken, numberOfTakes);
+ t.interrupt();
+ }
+
+ // Assert we can put exactly the numberOfPuts
+ public void assertCanPut(CallQueueManager<FakeCall> cq, int numberOfPuts,
+ int putAttempts) throws InterruptedException {
+
+ Putter putter = new Putter(cq, putAttempts, -1);
+ Thread t = new Thread(putter);
+ t.start();
+ t.join(100);
+
+ assertEquals(putter.callsAdded, numberOfPuts);
+ t.interrupt();
+ }
+
+
+ @Test
+ public void testCallQueueCapacity() throws InterruptedException {
+ manager = new CallQueueManager<FakeCall>(LinkedBlockingQueue.class, 10, "", null);
+
+ assertCanPut(manager, 10, 20); // Will stop at 10 due to capacity
+ }
+
+ @Test
+ public void testEmptyConsume() throws InterruptedException {
+ manager = new CallQueueManager<FakeCall>(LinkedBlockingQueue.class, 10, "", null);
+
+ assertCanTake(manager, 0, 1); // Fails since it's empty
+ }
+
+ @Test(timeout=60000)
+ public void testSwapUnderContention() throws InterruptedException {
+ manager = new CallQueueManager<FakeCall>(LinkedBlockingQueue.class, 5000, "", null);
+
+ ArrayList<Putter> producers = new ArrayList<Putter>();
+ ArrayList<Taker> consumers = new ArrayList<Taker>();
+
+ HashMap<Runnable, Thread> threads = new HashMap<Runnable, Thread>();
+
+ // Create putters and takers
+ for (int i=0; i < 50; i++) {
+ Putter p = new Putter(manager, -1, -1);
+ Thread pt = new Thread(p);
+ producers.add(p);
+ threads.put(p, pt);
+
+ pt.start();
+ }
+
+ for (int i=0; i < 20; i++) {
+ Taker t = new Taker(manager, -1, -1);
+ Thread tt = new Thread(t);
+ consumers.add(t);
+ threads.put(t, tt);
+
+ tt.start();
+ }
+
+ Thread.sleep(10);
+
+ assertTrue(manager.size() > 0);
+
+ for (int i=0; i < 5; i++) {
+ manager.swapQueue(LinkedBlockingQueue.class, 5000, "", null);
+ }
+
+ // Stop the producers
+ for (Putter p : producers) {
+ p.stop();
+ }
+
+ // Wait for consumers to wake up, then consume
+ Thread.sleep(2000);
+ assertEquals(0, manager.size());
+
+ // Ensure no calls were dropped
+ long totalCallsCreated = 0;
+ long totalCallsConsumed = 0;
+
+ for (Putter p : producers) {
+ totalCallsCreated += p.callsAdded;
+ threads.get(p).interrupt();
+ }
+ for (Taker t : consumers) {
+ totalCallsConsumed += t.callsTaken;
+ threads.get(t).interrupt();
+ }
+
+ assertEquals(totalCallsConsumed, totalCallsCreated);
+ }
+}
\ No newline at end of file