1 /*

     2  * Copyright 2004-2007 Sun Microsystems, Inc.  All Rights Reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.

     8  *

     9  * This code is distributed in the hope that it will be useful, but WITHOUT

    10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    11  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    12  * version 2 for more details (a copy is included in the LICENSE file that

    13  * accompanied this code).

    14  *

    15  * You should have received a copy of the GNU General Public License version

    16  * 2 along with this work; if not, write to the Free Software Foundation,

    17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    18  *

    19  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

    20  * CA 95054 USA or visit www.sun.com if you need additional information or

    21  * have any questions.

    22  */

    23 

    24 /*

    25  * @test

    26  * @bug 6190873

    27  * @summary Tests that thread creation can use a user-supplied Executor

    28  * @author Eamonn McManus

    29  * @run clean ExecutorTest

    30  * @run build ExecutorTest

    31  * @run main ExecutorTest

    32  */

    33 

    34 import java.lang.reflect.*;

    35 import java.net.MalformedURLException;

    36 

    37 import java.util.*;

    38 import java.util.concurrent.*;

    39 

    40 import javax.management.*;

    41 import javax.management.remote.*;

    42 

    43 /*

    44   When you create a JMXConnector client, you can supply a

    45   "fetch-notifications Executor", which is a

    46   java.util.concurrent.Executor that will be used each time the

    47   connector client wants to call RMIConnection.fetchNotifications.

    48   This is a hook that allows users to make that potentially-blocking

    49   call from within a thread pool or the like.  If you have very many

    50   connections, you can potentially share the work of

    51   fetchNotifications calls among a number of threads that is less than

    52   the number of connections, decreasing thread usage at the expense of

    53   increased latency.

    54 

    55   This test checks that the environment property does in fact work.

    56   It creates a connection without that property and ensures that

    57   notification sending does in fact work (with the default Executor).

    58   Then it creates a connection with the property set to an Executor

    59   that records how many times its execute method is invoked.

    60   Notifications are sent one by one and each time the test waits for

    61   the notification to be received.  This implies that

    62   fetchNotifications will be executed at least as many times as there

    63   are notifications.  Since each fetchNotifications call is supposed

    64   to happen as an Executor.execute call, if Executor.execute has been

    65   called fewer times then there were notifications, we know that the

    66   Executor is not being used correctly.

    67  */

    68 public class ExecutorTest {

    69     private static final String EXECUTOR_PROPERTY =

    70         "jmx.remote.x.fetch.notifications.executor";

    71     private static final String NOTIF_TYPE = "test.type";

    72 

    73     public static void main(String[] args) throws Exception {

    74         String lastfail = null;

    75         for (String proto : new String[] {"rmi", "iiop", "jmxmp"}) {

    76             JMXServiceURL url = new JMXServiceURL(proto, null, 0);

    77             JMXConnectorServer cs;

    78             MBeanServer mbs = MBeanServerFactory.newMBeanServer();

    79             try {

    80                 // Create server just to see if the protocol is supported

    81                 cs = JMXConnectorServerFactory.newJMXConnectorServer(url,

    82                                                                      null,

    83                                                                      mbs);

    84             } catch (MalformedURLException e) {

    85                 System.out.println();

    86                 System.out.println("Ignoring protocol: " + proto);

    87                 continue;

    88             }

    89             String fail;

    90             try {

    91                 fail = test(proto);

    92                 if (fail != null)

    93                     System.out.println("TEST FAILED: " + fail);

    94             } catch (Exception e) {

    95                 e.printStackTrace(System.out);

    96                 fail = e.toString();

    97             }

    98             if (lastfail == null)

    99                 lastfail = fail;

   100         }

   101         if (lastfail == null)

   102             return;

   103         System.out.println();

   104         System.out.println("TEST FAILED");

   105         throw new Exception("Test failed: " + lastfail);

   106     }

   107 

   108     private static enum TestType {NO_EXECUTOR, NULL_EXECUTOR, COUNT_EXECUTOR};

   109 

   110     private static String test(String proto) throws Exception {

   111         System.out.println();

   112         System.out.println("TEST: " + proto);

   113         ClassLoader myClassLoader =

   114             CountInvocationHandler.class.getClassLoader();

   115         ExecutorService wrappedExecutor = Executors.newCachedThreadPool();

   116         CountInvocationHandler executorHandler =

   117             new CountInvocationHandler(wrappedExecutor);

   118         Executor countExecutor = (Executor)

   119             Proxy.newProxyInstance(myClassLoader,

   120                                    new Class<?>[] {Executor.class},

   121                                    executorHandler);

   122 

   123         JMXServiceURL url = new JMXServiceURL(proto, null, 0);

   124 

   125         for (TestType test : TestType.values()) {

   126             Map<String, Executor> env = new HashMap<String, Executor>();

   127             switch (test) {

   128             case NO_EXECUTOR:

   129                 System.out.println("Test with no executor in env");

   130                 break;

   131             case NULL_EXECUTOR:

   132                 System.out.println("Test with null executor in env");

   133                 env.put(EXECUTOR_PROPERTY, null);

   134                 break;

   135             case COUNT_EXECUTOR:

   136                 System.out.println("Test with non-null executor in env");

   137                 env.put(EXECUTOR_PROPERTY, countExecutor);

   138                 break;

   139             }

   140             MBeanServer mbs = MBeanServerFactory.newMBeanServer();

   141             ObjectName emitName = new ObjectName("blah:type=Emitter");

   142             mbs.registerMBean(new Emitter(), emitName);

   143             JMXConnectorServer cs =

   144                 JMXConnectorServerFactory.newJMXConnectorServer(url, null, mbs);

   145             cs.start();

   146             JMXServiceURL addr = cs.getAddress();

   147             JMXConnector cc = JMXConnectorFactory.connect(addr, env);

   148             MBeanServerConnection mbsc = cc.getMBeanServerConnection();

   149             EmitterMBean emitter = (EmitterMBean)

   150                 MBeanServerInvocationHandler.newProxyInstance(mbsc,

   151                                                               emitName,

   152                                                               EmitterMBean.class,

   153                                                               false);

   154             SemaphoreListener listener = new SemaphoreListener();

   155             NotificationFilterSupport filter = new NotificationFilterSupport();

   156             filter.enableType(NOTIF_TYPE);

   157             mbsc.addNotificationListener(emitName, listener, filter, null);

   158             final int NOTIF_COUNT = 10;

   159             for (int i = 0; i < NOTIF_COUNT; i++) {

   160                 emitter.emit();

   161                 listener.await();

   162             }

   163             Thread.sleep(1);

   164             listener.checkUnavailable();

   165             System.out.println("Got notifications");

   166             cc.close();

   167             cs.stop();

   168             if (test == TestType.COUNT_EXECUTOR) {

   169                 Method m = Executor.class.getMethod("execute", Runnable.class);

   170                 Integer count = executorHandler.methodCount.get(m);

   171                 if (count == null || count < NOTIF_COUNT)

   172                     return "Incorrect method count for execute: " + count;

   173                 System.out.println("Executor was called enough times");

   174             }

   175         }

   176 

   177         wrappedExecutor.shutdown();

   178         return null;

   179     }

   180 

   181     /* Simple MBean that sends a notification every time we ask it to.  */

   182     public static interface EmitterMBean {

   183         public void emit();

   184     }

   185 

   186     public static class Emitter

   187             extends NotificationBroadcasterSupport implements EmitterMBean {

   188         public void emit() {

   189             sendNotification(new Notification(NOTIF_TYPE, this, seq++));

   190         }

   191 

   192         private long seq = 1;

   193     }

   194 

   195     /* Simple NotificationListener that allows you to wait until a

   196        notification has been received.  Since it uses a semaphore, you

   197        can wait either before or after the notification has in fact

   198        been received and it will work in either case.  */

   199     private static class SemaphoreListener implements NotificationListener {

   200         void await() throws InterruptedException {

   201             semaphore.acquire();

   202         }

   203 

   204         /* Ensure no extra notifications were received.  If we can acquire

   205            the semaphore, that means its release() method was called more

   206            times than its acquire() method, which means there were too

   207            many notifications.  */

   208         void checkUnavailable() throws Exception {

   209             if (semaphore.tryAcquire())

   210                 throw new Exception("Got extra notifications!");

   211         }

   212 

   213         public void handleNotification(Notification n, Object h) {

   214             semaphore.release();

   215         }

   216 

   217         private final Semaphore semaphore = new Semaphore(0);

   218     }

   219 

   220     /* Generic InvocationHandler that forwards all methods to a wrapped

   221        object, but counts for each method how many times it was invoked.  */

   222     private static class CountInvocationHandler implements InvocationHandler {

   223         final Map<Method, Integer> methodCount =

   224             new HashMap<Method, Integer>();

   225         private final Object wrapped;

   226 

   227         public CountInvocationHandler(Object wrapped) {

   228             this.wrapped = wrapped;

   229         }

   230 

   231         public Object invoke(Object proxy, Method method, Object[] args)

   232                 throws Throwable {

   233             synchronized (methodCount) {

   234                 Integer count = methodCount.get(method);

   235                 if (count == null)

   236                     count = 0;

   237                 methodCount.put(method, count + 1);

   238             }

   239             return method.invoke(wrapped, (Object[]) args);

   240         }

   241     }

   242 }