HDFS-5973. add DomainSocket#shutdown method. (cmccabe)

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

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/net/unix/DomainSocket.java

@@ -377,6 +377,22 @@ public class DomainSocket implements Closeable {
       Thread.currentThread().interrupt();
     }
   }
+  
+  /**
+   * Call shutdown(SHUT_RDWR) on the UNIX domain socket.
+   *
+   * @throws IOException
+   */
+  public void shutdown() throws IOException {
+    refCount.reference();
+    boolean exc = true;
+    try {
+      shutdown0(fd);
+      exc = false;
+    } finally {
+      unreference(exc);
+    }
+  }
 
   private native static void sendFileDescriptors0(int fd,
       FileDescriptor descriptors[],

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/net/unix/TestDomainSocket.java

@@ -34,6 +34,8 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.AfterClass;
 import org.junit.Assert;
@@ -41,7 +43,6 @@ import org.junit.Assume;
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
-
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.net.unix.DomainSocket.DomainChannel;
@@ -727,4 +728,38 @@ public class TestDomainSocket {
       tmp.close();
     }
   }
+
+  @Test(timeout=180000)
+  public void testShutdown() throws Exception {
+    final AtomicInteger bytesRead = new AtomicInteger(0);
+    final AtomicBoolean failed = new AtomicBoolean(false);
+    final DomainSocket[] socks = DomainSocket.socketpair();
+    Runnable reader = new Runnable() {
+      @Override
+      public void run() {
+        while (true) {
+          try {
+            int ret = socks[1].getInputStream().read();
+            if (ret == -1) return;
+            bytesRead.addAndGet(1);
+          } catch (IOException e) {
+            DomainSocket.LOG.error("reader error", e);
+            failed.set(true);
+            return;
+          }
+        }
+      }
+    };
+    Thread readerThread = new Thread(reader);
+    readerThread.start();
+    socks[0].getOutputStream().write(1);
+    socks[0].getOutputStream().write(2);
+    socks[0].getOutputStream().write(3);
+    Assert.assertTrue(readerThread.isAlive());
+    socks[0].shutdown();
+    readerThread.join();
+    Assert.assertFalse(failed.get());
+    Assert.assertEquals(3, bytesRead.get());
+    IOUtils.cleanup(null, socks);
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -87,6 +87,8 @@ Release 2.4.0 - UNRELEASED
     HDFS-5768. Consolidate the serialization code in DelegationTokenSecretManager
     (Haohui Mai via brandonli)
 
+    HDFS-5973. add DomainSocket#shutdown method (cmccabe)
+
   OPTIMIZATIONS
 
     HDFS-5790. LeaseManager.findPath is very slow when many leases need recovery