    1   /*
    2    * Licensed to the Apache Software Foundation (ASF) under one
    3    * or more contributor license agreements.  See the NOTICE file
    4    * distributed with this work for additional information
    5    * regarding copyright ownership.  The ASF licenses this file
    6    * to you under the Apache License, Version 2.0 (the
    7    * "License"); you may not use this file except in compliance
    8    * with the License.  You may obtain a copy of the License at
    9    *
    10   *   http://www.apache.org/licenses/LICENSE-2.0
    11   *
    12   * Unless required by applicable law or agreed to in writing,
    13   * software distributed under the License is distributed on an
    14   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    15   * KIND, either express or implied.  See the License for the
    16   * specific language governing permissions and limitations
    17   * under the License.
    18   */
    19  package org.apache.mina.transport.vmpipe;
    20
    21  import java.lang.management.ManagementFactory;
    22  import java.lang.management.ThreadInfo;
    23  import java.lang.management.ThreadMXBean;
    24  import java.util.concurrent.CountDownLatch;
    25  import java.util.concurrent.TimeUnit;
    26  import java.util.concurrent.atomic.AtomicReference;
    27
    28  import org.apache.mina.common.ConnectFuture;
    29  import org.apache.mina.common.IoAcceptor;
    30  import org.apache.mina.common.IoAcceptorConfig;
    31  import org.apache.mina.common.IoConnector;
    32  import org.apache.mina.common.IoHandlerAdapter;
    33  import org.apache.mina.common.IoSession;
    34  import org.apache.mina.common.ThreadModel;
    35
    36  import junit.framework.TestCase;
    37
    38  /**
    39   * @author Apache Mina Project (dev@mina.apache.org)
    40   * @version $Rev: $, $Date:  $
    41   */
    42  public class VmPipeSessionCrossCommunicationTest extends TestCase {
    43      public void testOneSessionTalkingBackAndForthDoesNotDeadlock() throws Exception {
    44          final VmPipeAddress address = new VmPipeAddress( 1 );
    45          final IoConnector connector = new VmPipeConnector();
    46          final AtomicReference<IoSession> c1 = new AtomicReference<IoSession>();
    47          final CountDownLatch latch = new CountDownLatch( 1 );
    48          final CountDownLatch messageCount = new CountDownLatch( 2 );
    49          IoAcceptor acceptor = new VmPipeAcceptor();
    50
    51          acceptor.bind( address, new IoHandlerAdapter() {
    52              @Override
    53              public void messageReceived( IoSession session, Object message ) throws Exception {
    54                  System.out.println( Thread.currentThread().getName() + ": " + message );
    55
    56                  if ( "start".equals( message ) ) {
    57                      session.write( "open new" );
    58                  } else if ( "re-use c1".equals( message ) ) {
    59                      session.write( "tell me something on c1 now" );
    60                  } else if ( ( (String) message ).startsWith( "please don't deadlock" ) ) {
    61                      messageCount.countDown();
    62                  } else {
    63                      fail( "unexpected message received " + message );
    64                  }
    65              }
    66          } );
    67
    68          connector.getDefaultConfig().setThreadModel( ThreadModel.MANUAL );
    69
    70          ConnectFuture future = connector.connect( address, new IoHandlerAdapter() {
    71              @Override
    72              public void messageReceived( IoSession session, Object message ) throws Exception {
    73                  System.out.println( Thread.currentThread().getName() + ": " + message );
    74
    75                  if ( "open new".equals( message ) ) {
    76                      System.out.println( "opening c2 from " + Thread.currentThread().getName() );
    77
    78                      ConnectFuture c2Future = connector.connect( address, new IoHandlerAdapter() {
    79                          @Override
    80                          public void sessionOpened( IoSession session ) throws Exception {
    81                              session.write( "re-use c1" );
    82                          }
    83
    84                          @Override
    85                          public void messageReceived( IoSession session, Object message ) throws Exception {
    86                              System.out.println( Thread.currentThread().getName() + ": " + message );
    87
    88                              if ( "tell me something on c1 now".equals( message ) ) {
    89                                  latch.countDown();
    90                                  c1.get().write( "please don't deadlock via c1" );
    91                              } else {
    92                                  fail( "unexpected message received " + message );
    93                              }
    94                          }
    95                      } );
    96
    97                      c2Future.join();
    98
    99                      latch.await();
    100
    101                     c2Future.getSession().write( "please don't deadlock via c2" );
    102                 } else {
    103                     fail( "unexpeced message received " + message );
    104                 }
    105             }
    106         } );
    107
    108         future.join();
    109
    110         c1.set( future.getSession() );
    111         c1.get().write( "start" );
    112
    113         ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
    114
    115         while ( !messageCount.await( 100, TimeUnit.MILLISECONDS ) ) {
    116             long[] threads = threadMXBean.findMonitorDeadlockedThreads();
    117
    118             if ( null != threads ) {
    119                 StringBuffer sb = new StringBuffer( 256 );
    120                 ThreadInfo[] infos = threadMXBean.getThreadInfo( threads, Integer.MAX_VALUE );
    121
    122                 for ( ThreadInfo info : infos ) {
    123                     sb.append( info.getThreadName() )
    124                         .append( " blocked on " )
    125                         .append( info.getLockName() )
    126                         .append( " owned by " )
    127                         .append( info.getLockOwnerName() )
    128                         .append( "\n" );
    129                 }
    130
    131                 for ( ThreadInfo info : infos ) {
    132                     sb.append( "\nStack for " ).append( info.getThreadName() ).append( "\n" );
    133                     for ( StackTraceElement element : info.getStackTrace() ) {
    134                         sb.append( "\t" ).append( element ).append( "\n" );
    135                     }
    136                 }
    137
    138                 fail( "deadlocked! \n" + sb );
    139             }
    140         }
    141
    142         ( (IoAcceptorConfig) acceptor.getDefaultConfig() ).setDisconnectOnUnbind( false );
    143         acceptor.unbindAll();
    144     }
    145 }