Handle many keep-alive clients and running slow code

[Luciano Macedo Rodrigues]

Hi,

I'm having a problem with MINA implementation. I intend to use it as an
TCP and UDP server in the same class, is there a problem with this?
My scenario is that we have some GPS devices that transmits data over
GPRS, with status and other sensors information, using a frame of at max
80 bytes, each byte defined by our protocol. I need to proccess the
information and verify if I need to send some commands to the device, so
I have to access a database. In each session, a keep a device object
with a Vector of the commands, so everytime I receive a message or some
message is sent, I call getDataToSend() to get the next command to send
and remove it from the Vector.

I'm using C3P0 for database pooling, so I don't have any overhead of
connecting to the database. But the code of processing a buffer (see
code above) that do the database code take about 400ms. The problem is
that I have many clients sending data simultaneosly, and I don't know if
that code is blocking something on the server. The device is configured
to resend the data after a timeout, and it's happening very often. I
also got lots of TCP Retransmission, and some TCP Out-of-Order and TCP
Dup Ack if I sniff the server.

So, how to handle many clients and run some slow code? I'm thinking of
some wait() and notify() operation, but I don't know if it's ok to use
it. A good idea would be a thread of IoHandler, but I don't know how to
implement it.

If you need more piece of code, just ask.

Thanks..

public class ServerSessionHandler implements IoHandler {

        private static DataSource ds;
        private static Utils utils;
        
    public ServerSessionHandler(DataSource ds, Utils utils) {
        this.ds = ds;
        this.utils = utils;
    }

        public void exceptionCaught(IoSession session, Throwable arg1)
throws Exception {
                System.err.println("exception caught");
                arg1.printStackTrace();
                session.close();
        }

        public void messageReceived(IoSession session, Object message)
throws Exception {
                ChipsatMessage buffer = (ChipsatMessage) message;
                
                Device device = (Device)session.getAttachment();
                if (device == null) { 
                        device = new Device();
                        session.setAttachment(device);
                }
                
                try {
                  long tbegin = System.currentTimeMillis();
                        // that's is a little bit l
                device.proccessBuffer(buffer.getBuffer(),ds,utils);


                long tend = System.currentTimeMillis();
                System.err.println("took "+(tend-tbegin)+"ms for
"+device.getIdEquipamento());
                
        
                Frame frame = device.getDataToSend();
                if (frame != null) {
                ChipsatMessage envio = new ChipsatMessage();
                envio.setBuffer(frame.getRet());
                session.write( envio );
                }
        
        }

        public void messageSent(IoSession session, Object message)
throws Exception {
                Device device = (Device)session.getAttachment();
                System.err.println("SND "+device.getIdEquipamento());
        
                Frame frame = device.getDataToSend();
                if (frame != null) {
                ChipsatMessage envio = new ChipsatMessage();
                envio.setBuffer(frame.getRet());
                envio.setIdEquipamento(frame.getEquipamento());
                session.write( envio );
                }
        }

        public void sessionClosed(IoSession session) throws Exception {
                Device device = (Device)session.getAttachment();
                device.clean();
        }

        public void sessionCreated(IoSession session) throws Exception {
                ProtocolCodecFactory codec = new
ChipsatProtocolCodecFactory();
                session.getFilterChain().addFirst(
                "protocolFilter", new ProtocolCodecFilter( codec ) );
                session.getFilterChain().addLast(
                "logger", new LoggingFilter() );
        
        IoSessionConfig cfg = session.getConfig();
        if( cfg instanceof SocketSessionConfig ) {
            SocketSessionConfig scfg = (SocketSessionConfig) cfg;
            scfg.setReceiveBufferSize(128);
            scfg.setSendBufferSize(320);
        }

        }

        public void sessionIdle(IoSession session, IdleStatus arg1)
throws Exception {
                session.close();
        }

        public void sessionOpened(IoSession session) throws Exception {
                session.setIdleTime( IdleStatus.BOTH_IDLE, 60);

        Device device = new Device();
        session.setAttachment(device);
                
        }
}


Luciano Macedo Rodrigues
[EMAIL PROTECTED]
Chipsat - Soluções em Rastreamento