6697180: JMX query results in java.io.IOException: Illegal state - also a deadlock can also be seen

Reviewed-by: emcmanus

src/share/classes/com/sun/jmx/remote/internal/ClientNotifForwarder.java

@@ -290,28 +290,6 @@ public abstract class ClientNotifForwarder {
 
         infoList.clear();
 
-        if (currentFetchThread == Thread.currentThread()) {
-            /* we do not need to stop the fetching thread, because this thread is
-               used to do restarting and it will not be used to do fetching during
-               the re-registering the listeners.*/
-            return tmp;
-        }
-
-        while (state == STARTING) {
-            try {
-                wait();
-            } catch (InterruptedException ire) {
-                IOException ioe = new IOException(ire.toString());
-                EnvHelp.initCause(ioe, ire);
-
-                throw ioe;
-            }
-        }
-
-        if (state == STARTED) {
-            setState(STOPPING);
-        }
-
         return tmp;
     }
 
@@ -353,8 +331,9 @@ public abstract class ClientNotifForwarder {
         beingReconnected = false;
         notifyAll();
 
-        if (currentFetchThread == Thread.currentThread()) {
-            // no need to init, simply get the id
+        if (currentFetchThread == Thread.currentThread() ||
+              state == STARTING || state == STARTED) { // doing or waiting reconnection
+              // only update mbeanRemovedNotifID
             try {
                 mbeanRemovedNotifID = addListenerForMBeanRemovedNotif();
             } catch (Exception e) {
@@ -366,12 +345,23 @@ public abstract class ClientNotifForwarder {
                     logger.trace("init", msg, e);
                 }
             }
-        } else if (listenerInfos.length > 0) { // old listeners re-registered
-            init(true);
-        } else if (infoList.size() > 0) {
-            // but new listeners registered during reconnection
-            init(false);
-        }
+        } else {
+              while (state == STOPPING) {
+                  try {
+                      wait();
+                  } catch (InterruptedException ire) {
+                      IOException ioe = new IOException(ire.toString());
+                      EnvHelp.initCause(ioe, ire);
+                      throw ioe;
+                  }
+              }
+
+              if (listenerInfos.length > 0) { // old listeners are re-added
+                  init(true); // not update clientSequenceNumber
+              } else if (infoList.size() > 0) { // only new listeners added during reconnection
+                  init(false); // need update clientSequenceNumber
+              }
+          }
     }
 
     public synchronized void terminate() {
@@ -486,6 +476,15 @@ public abstract class ClientNotifForwarder {
             if (nr == null || shouldStop()) {
                 // tell that the thread is REALLY stopped
                 setState(STOPPED);
+
+                try {
+                      removeListenerForMBeanRemovedNotif(mbeanRemovedNotifID);
+                } catch (Exception e) {
+                    if (logger.traceOn()) {
+                        logger.trace("NotifFetcher-run",
+                                "removeListenerForMBeanRemovedNotif", e);
+                    }
+                }
             } else {
                 executor.execute(this);
             }

test/javax/management/remote/mandatory/connection/MultiThreadDeadLockTest.java

+/*
+ * To change this template, choose Tools | Templates
+ * and open the template in the editor.
+ */
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.net.Socket;
+import java.rmi.server.RMIClientSocketFactory;
+import java.util.HashMap;
+import javax.management.MBeanServer;
+import javax.management.MBeanServerFactory;
+import javax.management.Notification;
+import javax.management.NotificationBroadcasterSupport;
+import javax.management.NotificationListener;
+import javax.management.ObjectName;
+import javax.management.remote.JMXConnector;
+import javax.management.remote.JMXConnectorFactory;
+import javax.management.remote.JMXConnectorServer;
+import javax.management.remote.JMXConnectorServerFactory;
+import javax.management.remote.JMXServiceURL;
+import javax.management.remote.rmi.RMIConnectorServer;
+
+/*
+ * @test
+ * @bug 6697180
+ * @summary test on a client notification deadlock.
+ * @author Shanliang JIANG
+ * @run clean MultiThreadDeadLockTest
+ * @run build MultiThreadDeadLockTest
+ * @run main MultiThreadDeadLockTest
+ */
+
+public class MultiThreadDeadLockTest {
+
+    private static long serverTimeout = 500L;
+
+    public static void main(String[] args) throws Exception {
+        print("Create the MBean server");
+        MBeanServer mbs = MBeanServerFactory.createMBeanServer();
+
+        print("Initialize environment map");
+        HashMap env = new HashMap();
+
+        print("Specify a client socket factory to control socket creation.");
+        env.put(RMIConnectorServer.RMI_CLIENT_SOCKET_FACTORY_ATTRIBUTE,
+                clientFactory);
+
+        print("Specify a server idle timeout to make a server close an idle connection.");
+        env.put("jmx.remote.x.server.connection.timeout", serverTimeout);
+
+        print("Disable client heartbeat.");
+        env.put("jmx.remote.x.client.connection.check.period", 0);
+
+        env.put("jmx.remote.x.notification.fetch.timeout", serverTimeout);
+
+        print("Create an RMI server");
+        JMXServiceURL url = new JMXServiceURL("rmi", null, 0);
+        JMXConnectorServer server =
+                JMXConnectorServerFactory.newJMXConnectorServer(url, env, mbs);
+        server.start();
+
+        url = server.getAddress();
+
+        print("Create jmx client on "+url);
+        StateMachine.setState(CREATE_SOCKET); // allow to create client socket
+        client = JMXConnectorFactory.connect(url, env);
+        Thread.sleep(100);
+
+        totoName = new ObjectName("default:name=toto");
+        mbs.registerMBean(toto, totoName);
+        print("Register the mbean: " + totoName);
+
+        print("Add listener to toto MBean");
+        client.getMBeanServerConnection().addNotificationListener(
+                totoName, myListener, null, null);
+        Thread.sleep(10);
+
+        print("send notif, listener will block the fetcher");
+        toto.sendNotif();
+        Thread.sleep(100);
+
+        StateMachine.setState(NO_OP);
+
+        print("Sleep 3 times of server idle timeout: "+serverTimeout+
+                ", the sever should close the idle connection.");
+        Thread.sleep(serverTimeout*3);
+
+        print("start the user thread to call mbean method, it will get IOexception" +
+                " and start the reconnection, the socket factory will block the" +
+                " socket creation.");
+        UserThread ut = new UserThread();
+        ut.start();
+        Thread.sleep(10);
+
+        print("Free the listener, the fetcher will get IO and makes " +
+                "a deadlock if the bug is not fixed.");
+        StateMachine.setState(FREE_LISTENER);
+        Thread.sleep(100);
+
+        print("Allow to create new socket for the reconnection");
+        StateMachine.setState(CREATE_SOCKET);
+
+        print("Check whether the user thread gets free to call the mbean.");
+        if (!ut.waitDone(5000)) {
+            throw new RuntimeException("Possible deadlock!");
+        }
+
+        print("Remove the listener.");
+        client.getMBeanServerConnection().removeNotificationListener(
+                totoName, myListener, null, null);
+        Thread.sleep(serverTimeout*3);
+
+        print("\nWell passed, bye!");
+
+        client.close();
+        Thread.sleep(10);
+        server.stop();
+    }
+
+    private static ObjectName totoName = null;
+    private static JMXConnector client;
+
+    public static class UserThread extends Thread {
+        public UserThread() {
+            setDaemon(true);
+        }
+
+        public void run() {
+            try {
+                client.getMBeanServerConnection().invoke(
+                        totoName, "allowReturn", null, null);
+            } catch (Exception e) {
+                throw new Error(e);
+            }
+
+            synchronized(UserThread.class) {
+                done = true;
+                UserThread.class.notify();
+            }
+        }
+
+        public boolean waitDone(long timeout) {
+            synchronized(UserThread.class) {
+                if(!done) {
+                    try {
+                        UserThread.class.wait(timeout);
+                    } catch (Exception e) {
+                        throw new Error(e);
+                    }
+                }
+            }
+            return done;
+        }
+
+        private boolean done = false;
+    }
+
+    public static interface TotoMBean {
+        public void allowReturn();
+    }
+
+    public static class Toto extends NotificationBroadcasterSupport
+            implements TotoMBean {
+
+        public void allowReturn() {
+            enter("allowReturn");
+
+            leave("allowReturn");
+        }
+
+        public void sendNotif() {
+            enter("sendNotif");
+
+            sendNotification(new Notification("Toto", totoName, 0));
+
+            leave("sendNotif");
+        }
+    }
+    private static Toto toto = new Toto();
+
+    public static NotificationListener myListener = new NotificationListener() {
+        public void handleNotification(Notification notification, Object handback) {
+            enter("handleNotification");
+
+            StateMachine.waitState(FREE_LISTENER);
+
+            leave("handleNotification");
+        }
+    };
+
+    public static class RMIClientFactory
+            implements RMIClientSocketFactory, Serializable {
+
+        public Socket createSocket(String host, int port) throws IOException {
+            enter("createSocket");
+            //print("Calling createSocket(" + host + " " + port + ")");
+
+            StateMachine.waitState(CREATE_SOCKET);
+            Socket s = new Socket(host, port);
+            leave("createSocket");
+
+            return s;
+        }
+    }
+    private static RMIClientFactory clientFactory = new RMIClientFactory();
+
+    private static int CREATE_SOCKET = 1;
+    private static int FREE_LISTENER = 3;
+    private static int NO_OP = 0;
+
+    public static class StateMachine {
+
+        private static int state = NO_OP;
+        private static int[] lock = new int[0];
+
+        public static void waitState(int s) {
+            synchronized (lock) {
+                while (state != s) {
+                    try {
+                        lock.wait();
+                    } catch (InterruptedException ire) {
+                        // should not
+                        throw new Error(ire);
+                    }
+                }
+            }
+        }
+
+        public static int getState() {
+            synchronized (lock) {
+                return state;
+            }
+        }
+
+        public static void setState(int s) {
+            synchronized (lock) {
+                state = s;
+                lock.notifyAll();
+            }
+        }
+    }
+
+    private static void print(String m) {
+        System.out.println(m);
+    }
+
+    private static void enter(String m) {
+        System.out.println("\n---Enter the method " + m);
+    }
+
+    private static void leave(String m) {
+        System.out.println("===Leave the method: " + m);
+    }
+}
+