[DIRMINA-1173] Apache mina 2.2.1 threads blocking on ConnectFuture.awaitUninterruptibly() for ever

• Type: Bug
• Status: Open
• Priority: Critical
• Resolution: Unresolved
• Affects Version/s: 2.2.1
• Component/s: Core
• Flags: Important
• Assignee: Unassigned
• Reporter: KMVS
• Created: 26/Apr/23 09:43
• Updated: 4 days ago 23:53

Description

Hi All,

I have attached thread dump too for analysis.

I have migrated from 2.0.21 to 2.023 for solving CVE, i have seen thread blocking issue, So used latest mina verstion 2.2.1 but still threads were blocked here is the sample code.Thread hungs at awaitUninterruptibly.  Once this issue comes  in sub sequest launches nothing will work all threads will be blocked forever,i have to restart the process to make it work. For single thread working fine,if i start 50-100 threads this thread blocking issue will surface.I am using the same NioSocketConnector across the threads.

Here is an scenario, I have one gate way IP which i connect initially ,this gate way will return list of ips.
now i will close the previous IOsession and will create a nio connection with the first ip in the list.if it did n't work then again i will try to 
connect to next ip etc..
All these are in the critical section. i.e i will acquire a lock and then after successful connection i will release the lock.
But problem i have noticed is with the awaitUninterruptibly().
I have a state machine too. So connection set up is in one thread and response processing is in another thread.

 

Thread 1:

 Public class g10CaptureService
 {
 

private static final ProtocolCodecFilter probeCodecFilter = new ProtocolCodecFilter(new ProbeCodecFactory(G10Message.class));

    private static final ExecutorFilter executorFilter = new ExecutorFilter(16,32);

    private static final G10GPBMessageIoFilter gpbMessageFilter = new G10GPBMessageIoFilter(G10ParserContextFactory.getG10ParsingAndEncodingInstance());

 static

{ initConnectors() }

protected static void initConnectors()

{

StateMachine stateMachine = StateMachineFactory.getInstance(IoHandlerTransition.class).create(
                G10MinaClient.CONNECTED, new G10MinaClient(processor));

        IoHandler ioHandler = new StateMachineProxyBuilder().setStateContextLookup(
                new IoSessionStateContextLookup(new StateContextFactory() {
                    @Override
                    public StateContext create()

{                         final G10StateContext stateContext = new G10StateContext();                         stateContext.setStartedTime(new Date());                         return stateContext;                     }

                })).create(IoHandler.class, stateMachine);
                
                
//Global connector across the system, i.e multiple threads uses the same connector.            

NioSocketConnector connector = new NioSocketConnector();
        connector.getFilterChain().addLast("LoggingFilter", G10CaptureService.loggingFilter);
        connector.getFilterChain().addLast("codecFilter", G10CaptureService.probeCodecFilter);
        connector.getFilterChain().addLast("executorFilter", G10CaptureService.executorFilter);
        connector.getFilterChain().addLast("gpbMessageFilter", G10CaptureService.gpbMessageFilter);
        connector.getFilterChain().addLast("keepAliveFilter", G10CaptureService.keepAliveFilter);
        connector.setHandler(ioHandler);
}

        
public void StartRecordCapture()
{

connectionLock.lock();
try
{
ConnectFuture primaryConnectFuture = connector.connect(primaryAddress, initializer);
//hungs forever if the no. of threads are more than 30
primaryConnectFuture.awaitUninterruptibly();//no time out specified
if (!primaryConnectFuture.isConnected()) 
{
               
                    if (handleIOException(searchExpression, captureHandler))

{                         return;                     }

                    LOG.info("{} Apache mina connection setup time out happend.",
                    handleConnectionFailed(primaryAddress, captureHandler, "Primary IP connection timeout");
                    return;
}

}catch(Exception e)
{

}finally

{ connectionLock.unLock(); }

}        

Thread 2:

public void processGatewaySrQueryResponseSuccess(G10StateContext context, IoSession session, GatewaySrQueryResponse response) {

//get the g10CaptureService  from the Thread1 IOsession 

if(response.getIpPortCount() > 0) {
        try {
        List<IpAddrPort> msgIpPorts = response.getIpPortList();
        List<InetSocketAddress> probeIpPorts = new ArrayList<InetSocketAddress>();
        for (IpAddrPort ipp : msgIpPorts)

{            InetSocketAddress inetSockAddr = new InetSocketAddress(InetAddress.getByAddress(ipp.getIp().toByteArray()), ipp.getPort());         probeIpPorts.add(inetSockAddr);         }

        g10CaptureService.setProbeIps(probeIpPorts);
        } catch (Exception ex) {
        LOG.error("{} Exception occured while creating probe IP/Port.Exception:{}", getLogStr(session), ex);
        }

      context.addGeoBladeGwSession(session);        
 
        g10CaptureService.closeConnectionForSession(session); //cose thread1 io session
        g10CaptureService.setIsGeoBladeIdlSession(true);
        g10CaptureService.startRecordCapture(searchExpression, captureHandler, captureId); //connect to the probe again using 
        same thread 1 instance with different ip
}