avro-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cutt...@apache.org
Subject svn commit: r1384514 - in /avro/trunk: CHANGES.txt lang/java/ipc/src/main/java/org/apache/avro/ipc/NettyTransceiver.java lang/java/ipc/src/test/java/org/apache/avro/ipc/TestNettyTransceiverWhenServerStops.java
Date Thu, 13 Sep 2012 20:34:48 GMT
Author: cutting
Date: Thu Sep 13 20:34:48 2012
New Revision: 1384514

URL: http://svn.apache.org/viewvc?rev=1384514&view=rev
Log:
AVRO-1154. Java: Fix NettyTransciever to not hang when the server is stopped.  Contributed
by Bruno Dumon & Karel Vervaeke.

Added:
    avro/trunk/lang/java/ipc/src/test/java/org/apache/avro/ipc/TestNettyTransceiverWhenServerStops.java
  (with props)
Modified:
    avro/trunk/CHANGES.txt
    avro/trunk/lang/java/ipc/src/main/java/org/apache/avro/ipc/NettyTransceiver.java

Modified: avro/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/avro/trunk/CHANGES.txt?rev=1384514&r1=1384513&r2=1384514&view=diff
==============================================================================
--- avro/trunk/CHANGES.txt (original)
+++ avro/trunk/CHANGES.txt Thu Sep 13 20:34:48 2012
@@ -52,6 +52,9 @@ Avro 1.7.2 (unreleased)
     AVRO-851. Java: Fix a bug in GenericData#toString() when escaping
     characters. (Jeff Mesnil via cutting)
 
+    AVRO-1154. Java: Fix NettyTransciever to not hang when the server
+    is stopped. (Bruno Dumon via cutting)
+
 Avro 1.7.1 (16 July 2012)
 
   NEW FEATURES

Modified: avro/trunk/lang/java/ipc/src/main/java/org/apache/avro/ipc/NettyTransceiver.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/ipc/src/main/java/org/apache/avro/ipc/NettyTransceiver.java?rev=1384514&r1=1384513&r2=1384514&view=diff
==============================================================================
--- avro/trunk/lang/java/ipc/src/main/java/org/apache/avro/ipc/NettyTransceiver.java (original)
+++ avro/trunk/lang/java/ipc/src/main/java/org/apache/avro/ipc/NettyTransceiver.java Thu Sep
13 20:34:48 2012
@@ -28,8 +28,8 @@ import java.util.concurrent.ConcurrentHa
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.avro.Protocol;
 import org.apache.avro.ipc.NettyTransportCodec.NettyDataPack;
@@ -76,6 +76,10 @@ public class NettyTransceiver extends Tr
   private final ClientBootstrap bootstrap;
   private final InetSocketAddress remoteAddr;
   
+  volatile ChannelFuture channelFuture;
+  volatile boolean stopping;
+  private final Object channelFutureLock = new Object();
+
   /**
    * Read lock must be acquired whenever using non-final state.
    * Write lock must be acquired whenever modifying state.
@@ -89,6 +93,7 @@ public class NettyTransceiver extends Tr
     connectTimeoutMillis = 0L;
     bootstrap = null;
     remoteAddr = null;
+    channelFuture = null;
   }
 
   /**
@@ -242,14 +247,24 @@ public class NettyTransceiver extends Tr
       stateLock.writeLock().lock();
       try {
         if (!isChannelReady(channel)) {
+          synchronized(channelFutureLock) {
+            if (!stopping) {
           LOG.debug("Connecting to " + remoteAddr);
-          ChannelFuture channelFuture = bootstrap.connect(remoteAddr);
+              channelFuture = bootstrap.connect(remoteAddr);
+            }
+          }
+          if (channelFuture != null) {
           channelFuture.awaitUninterruptibly(connectTimeoutMillis);
+
+            synchronized(channelFutureLock) {
           if (!channelFuture.isSuccess()) {
             throw new IOException("Error connecting to " + remoteAddr, 
                 channelFuture.getCause());
           }
           channel = channelFuture.getChannel();
+              channelFuture = null;
+            }
+          }
         }
       } finally {
         // Downgrade to read lock:
@@ -280,6 +295,12 @@ public class NettyTransceiver extends Tr
     Channel channelToClose = null;
     Map<Integer, Callback<List<ByteBuffer>>> requestsToCancel = null;
     boolean stateReadLockHeld = stateLock.getReadHoldCount() != 0;
+
+    synchronized(channelFutureLock) {
+        if (stopping && channelFuture != null) {
+           channelFuture.cancel();
+        }
+    }
     if (stateReadLockHeld) {
       stateLock.readLock().unlock();
     }
@@ -350,6 +371,7 @@ public class NettyTransceiver extends Tr
   public void close() {
     try {
       // Close the connection:
+      stopping = true;
       disconnect(true, true, null);
     } finally {
       // Shut down all thread pools to exit.

Added: avro/trunk/lang/java/ipc/src/test/java/org/apache/avro/ipc/TestNettyTransceiverWhenServerStops.java
URL: http://svn.apache.org/viewvc/avro/trunk/lang/java/ipc/src/test/java/org/apache/avro/ipc/TestNettyTransceiverWhenServerStops.java?rev=1384514&view=auto
==============================================================================
--- avro/trunk/lang/java/ipc/src/test/java/org/apache/avro/ipc/TestNettyTransceiverWhenServerStops.java
(added)
+++ avro/trunk/lang/java/ipc/src/test/java/org/apache/avro/ipc/TestNettyTransceiverWhenServerStops.java
Thu Sep 13 20:34:48 2012
@@ -0,0 +1,139 @@
+/**
+ * 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.avro.ipc;
+
+import org.apache.avro.ipc.specific.SpecificRequestor;
+import org.apache.avro.ipc.specific.SpecificResponder;
+import org.apache.avro.test.Mail;
+import org.apache.avro.test.Message;
+import org.junit.Test;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.fail;
+
+public class TestNettyTransceiverWhenServerStops {
+  @Test
+    public void testNettyTransceiverWhenServerStops() throws Exception {
+    Mail mailService = new TestNettyServer.MailImpl();
+    Responder responder = new SpecificResponder(Mail.class, mailService);
+    NettyServer server = new NettyServer(responder, new InetSocketAddress(0));
+    server.start();
+
+    int serverPort = server.getPort();
+
+    final NettyTransceiver transceiver = new NettyTransceiver(new InetSocketAddress(serverPort),
60000L);
+    final Mail mail = SpecificRequestor.getClient(Mail.class, transceiver);
+
+    final AtomicInteger successes = new AtomicInteger();
+    final AtomicInteger failures = new AtomicInteger();
+    final AtomicBoolean quitOnFailure = new AtomicBoolean();
+    List<Thread> threads = new ArrayList<Thread>();
+
+    // Start a bunch of client threads that use the transceiver to send messages
+    for (int i = 0; i < 100; i++) {
+      Thread thread = new Thread(new Runnable() {
+          @Override
+            public void run() {
+            while (true) {
+              try {
+                mail.send(createMessage());
+                successes.incrementAndGet();
+              } catch (Exception e) {
+                failures.incrementAndGet();
+                if (quitOnFailure.get()) {
+                  return;
+                }
+              }
+            }
+          }
+        });
+      threads.add(thread);
+      thread.start();
+    }
+
+    // Be sure the threads are running: wait until we get a good deal of successes
+    while (successes.get() < 10000) {
+      Thread.sleep(50);
+    }
+
+    // Now stop the server
+    server.close();
+
+    // Server is stopped: successes should not increase anymore: wait until we're in that
situation
+    while (true) {
+      int previousSuccesses = successes.get();
+      Thread.sleep(500);
+      if (previousSuccesses == successes.get()) {
+        break;
+      }
+    }
+
+    // Start the server again
+    server.start();
+
+    // This part of the test is not solved by the current patch: it shows that when you stop/start
+    // a server, the client requests don't continue immediately but will stay blocked until
the timeout
+    // passed to the NettyTransceiver has passed (IIUC)
+    long now = System.currentTimeMillis();
+    /*
+      System.out.println("Waiting on requests to continue");
+      int previousSuccesses = successes.get();
+      while (true) {
+      Thread.sleep(500);
+      if (successes.get() > previousSuccesses) {
+      break;
+      }
+      if (System.currentTimeMillis() - now > 5000) {
+      System.out.println("FYI: requests don't continue immediately...");
+      break;
+      }
+      }
+    */
+
+    // Stop our client, we would expect this to go on immediately
+    System.out.println("Stopping transceiver");
+    quitOnFailure.set(true);
+    now = System.currentTimeMillis();
+    transceiver.close(); // Without the patch, this close seems to hang forever
+
+    // Wait for all threads to quit
+    for (Thread thread : threads) {
+      thread.join();
+    }
+    if (System.currentTimeMillis() - now > 10000) {
+      fail("Stopping NettyTransceiver and waiting for client threads to quit took too long.");
+    } else {
+      System.out.println("Stopping NettyTransceiver and waiting for client threads to quit
took "
+                         + (System.currentTimeMillis() - now) + " ms");
+    }
+  }
+
+  private Message createMessage() {
+    Message msg = Message.newBuilder().
+      setTo("wife").
+      setFrom("husband").
+      setBody("I love you!").
+      build();
+    return msg;
+  }
+}

Propchange: avro/trunk/lang/java/ipc/src/test/java/org/apache/avro/ipc/TestNettyTransceiverWhenServerStops.java
------------------------------------------------------------------------------
    svn:eol-style = native



Mime
View raw message