[DIRMINA-682] We need a better documentation for the ExecutorFilter [was :Writing more than one message will block until the MessageReceived as been fully proceced] - ASF JIRA

Details

Type: Improvement
Status: Resolved
Priority: Critical
Resolution: Not A Problem
Affects Version/s: 2.0.0-M4
Fix Version/s: 2.0.8
Component/s: None
Labels:
None

Description

When a message generates mor ethan one response, then the written responses will be sent to the client only when the initial message has been totally processed.

Suppose that we receive one message M, it will be handled by a IoProcessor in the process() method, go through the chain to the IoHandler.MessageReceive() method. Now, if one want to write more than one response (session.write( R )), then those responses will be enqueued until we are back to the process() method.

The issue is due to the fact that the write is done using the IoProcessor associated to the current session, leading to a problem : we can't ask the IoProcessor instance to unqueue the written message until it is done with the current processing( it's running in one single thread).

The consequences are painfull :

if one tries to write two responses, waiting for the first responses to be written, this will end with a DeadLock, as we are waiting on the processor we are holding
if we don't care about waiting for the write to be done, then all the responses will be enqueued and stored in memory, until the IoProcessor exit from the read processing and start processing the writes, leading to OOM Exception

One solution would be to have to sets of IoProcessors, one for the read, and one for the writes. Or to pick a random Processor to process the writes, as soon as the processor is not the same as the one processing the reads.

Here is a sample exhibiting the problem. Just launch it, and use 'telnet localhost 8080' in a console, type something, it should write twice the typed message, but it just generates an exception - see further - and write back the message once. Removing the wait will work, but the messages will be sent only when the read has been processed in the AbstractPollingIoProcessor.process(T session) method :

    /**
     * Deal with session ready for the read or write operations, or both.
     */
    private void process(T session) {
        // Process Reads
        if (isReadable(session) && !session.isReadSuspended()) {
            read(session);
        }

        // Process writes
        if (isWritable(session) && !session.isWriteSuspended()) {
            scheduleFlush(session);
        }
    }

The sample code :

package org.apache.mina.real.life;

import java.net.InetSocketAddress;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

/**
 * (<b>Entry point</b>) Echo server
 *
 * @author The Apache MINA Project (dev@mina.apache.org)
 */
public class Main {
    private static class EchoProtocolHandler extends IoHandlerAdapter {
        public void messageReceived(IoSession session, Object message)
                throws Exception {
            System.out.println(new String(((IoBuffer)message).array()));
            // Write the received data back to remote peer
            WriteFuture wf = session.write(((IoBuffer) message).duplicate());
            
            // Here, we will get a Deadlock detection
            wf.awaitUninterruptibly();
            
            // Do a second write
            session.write(((IoBuffer) message).duplicate());
        }
    }    
    /** Choose your favorite port number. */
    private static final int PORT = 8080;

    public static void main(String[] args) throws Exception {
        SocketAcceptor acceptor = new NioSocketAcceptor();
        
        // Add a logging filter
        acceptor.getFilterChain().addLast( "Logger", new LoggingFilter() );
        
        // Bind
        acceptor.setHandler(new EchoProtocolHandler());
        acceptor.bind(new InetSocketAddress(PORT));

        System.out.println("Listening on port " + PORT);
    }
}

The exception :

Listening on port 8080
[20:08:17] INFO [org.apache.mina.filter.logging.LoggingFilter] - CREATED
[20:08:17] INFO [org.apache.mina.filter.logging.LoggingFilter] - OPENED
[20:08:19] INFO [org.apache.mina.filter.logging.LoggingFilter] - RECEIVED: HeapBuffer[pos=0 lim=6 cap=2048: 74 65 73 74 0D 0A]
test

[20:08:30] WARN [org.apache.mina.filter.logging.LoggingFilter] - EXCEPTION :
java.lang.IllegalStateException: DEAD LOCK: IoFuture.await() was invoked from an I/O processor thread. Please use IoFutureListener or configure a proper thread model alternatively.
at org.apache.mina.core.future.DefaultIoFuture.checkDeadLock(DefaultIoFuture.java:235)
at org.apache.mina.core.future.DefaultIoFuture.await0(DefaultIoFuture.java:203)
at org.apache.mina.core.future.DefaultIoFuture.awaitUninterruptibly(DefaultIoFuture.java:131)
at org.apache.mina.core.future.DefaultWriteFuture.awaitUninterruptibly(DefaultWriteFuture.java:114)
at org.apache.mina.real.life.Main$EchoProtocolHandler.messageReceived(Main.java:45)
at org.apache.mina.core.filterchain.DefaultIoFilterChain$TailFilter.messageReceived(DefaultIoFilterChain.java:722)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.callNextMessageReceived(DefaultIoFilterChain.java:434)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.access$1200(DefaultIoFilterChain.java:48)
at org.apache.mina.core.filterchain.DefaultIoFilterChain$EntryImpl$1.messageReceived(DefaultIoFilterChain.java:802)
at org.apache.mina.filter.logging.LoggingFilter.messageReceived(LoggingFilter.java:178)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.callNextMessageReceived(DefaultIoFilterChain.java:434)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.access$1200(DefaultIoFilterChain.java:48)
at org.apache.mina.core.filterchain.DefaultIoFilterChain$EntryImpl$1.messageReceived(DefaultIoFilterChain.java:802)
at org.apache.mina.core.filterchain.IoFilterAdapter.messageReceived(IoFilterAdapter.java:120)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.callNextMessageReceived(DefaultIoFilterChain.java:434)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.fireMessageReceived(DefaultIoFilterChain.java:426)
at org.apache.mina.core.polling.AbstractPollingIoProcessor.read(AbstractPollingIoProcessor.java:604)
at org.apache.mina.core.polling.AbstractPollingIoProcessor.process(AbstractPollingIoProcessor.java:564)
at org.apache.mina.core.polling.AbstractPollingIoProcessor.process(AbstractPollingIoProcessor.java:553)
at org.apache.mina.core.polling.AbstractPollingIoProcessor.access$400(AbstractPollingIoProcessor.java:57)
at org.apache.mina.core.polling.AbstractPollingIoProcessor$Processor.run(AbstractPollingIoProcessor.java:892)
at org.apache.mina.util.NamePreservingRunnable.run(NamePreservingRunnable.java:65)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:619)
[20:08:30] WARN [org.apache.mina.real.life.Main$EchoProtocolHandler] - EXCEPTION, please implement org.apache.mina.real.life.Main$EchoProtocolHandler.exceptionCaught() for proper handling:
java.lang.IllegalStateException: DEAD LOCK: IoFuture.await() was invoked from an I/O processor thread. Please use IoFutureListener or configure a proper thread model alternatively.
at org.apache.mina.core.future.DefaultIoFuture.checkDeadLock(DefaultIoFuture.java:235)
at org.apache.mina.core.future.DefaultIoFuture.await0(DefaultIoFuture.java:203)
at org.apache.mina.core.future.DefaultIoFuture.awaitUninterruptibly(DefaultIoFuture.java:131)
at org.apache.mina.core.future.DefaultWriteFuture.awaitUninterruptibly(DefaultWriteFuture.java:114)
at org.apache.mina.real.life.Main$EchoProtocolHandler.messageReceived(Main.java:45)
at org.apache.mina.core.filterchain.DefaultIoFilterChain$TailFilter.messageReceived(DefaultIoFilterChain.java:722)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.callNextMessageReceived(DefaultIoFilterChain.java:434)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.access$1200(DefaultIoFilterChain.java:48)
at org.apache.mina.core.filterchain.DefaultIoFilterChain$EntryImpl$1.messageReceived(DefaultIoFilterChain.java:802)
at org.apache.mina.filter.logging.LoggingFilter.messageReceived(LoggingFilter.java:178)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.callNextMessageReceived(DefaultIoFilterChain.java:434)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.access$1200(DefaultIoFilterChain.java:48)
at org.apache.mina.core.filterchain.DefaultIoFilterChain$EntryImpl$1.messageReceived(DefaultIoFilterChain.java:802)
at org.apache.mina.core.filterchain.IoFilterAdapter.messageReceived(IoFilterAdapter.java:120)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.callNextMessageReceived(DefaultIoFilterChain.java:434)
at org.apache.mina.core.filterchain.DefaultIoFilterChain.fireMessageReceived(DefaultIoFilterChain.java:426)
at org.apache.mina.core.polling.AbstractPollingIoProcessor.read(AbstractPollingIoProcessor.java:604)
at org.apache.mina.core.polling.AbstractPollingIoProcessor.process(AbstractPollingIoProcessor.java:564)
at org.apache.mina.core.polling.AbstractPollingIoProcessor.process(AbstractPollingIoProcessor.java:553)
at org.apache.mina.core.polling.AbstractPollingIoProcessor.access$400(AbstractPollingIoProcessor.java:57)
at org.apache.mina.core.polling.AbstractPollingIoProcessor$Processor.run(AbstractPollingIoProcessor.java:892)
at org.apache.mina.util.NamePreservingRunnable.run(NamePreservingRunnable.java:65)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:619)
[20:08:30] INFO [org.apache.mina.filter.logging.LoggingFilter] - SENT: HeapBuffer[pos=0 lim=6 cap=2048: 74 65 73 74 0D 0A]

Attachments

Issue Links

relates to

DIRMINA-714 Packet sequence is unordered in multi thread.

Closed

We need a better documentation for the ExecutorFilter [was :Writing more than one message will block until the MessageReceived as been fully proceced]

Details

Description

Attachments

Issue Links

Activity

People

Dates