Graceful shutdown

[Liche]

I'm writing a server in netty for the first time. The protocol I'm
implementing allows clients to log in and open sessions, with which they
will send requests which will alter our database state. There are additional
requirements, which are particularly relevant when the server is shut down:
a) no incomplete messages may be sent; b) no request may be processed (and
database state altered) without a reply being sent to the user; c) greeting
messages must be sent to newly-connected users.

I have a draft implementation which shuts down gracefully with respect to a)
and b), but cannot get c) working. It is entirely possible that my approach
is wholly wrong and needs to be redone! I've basically solved it by having a
separate MySession object which retains the logged-in state, client
identification, and a lock. The lock is held during the processing of a
request, and released once the response has finished being sent. The session
cannot be terminated while the lock is held, and when the session terminates
it closes its parent channel.

The problem comes when the shutdown is invoked while a client is being sent
a greeting in channelConnected(...). In order to avoid either incomplete
messages being sent and/or ClosedChannelException being thrown (i.e. channel
closed by one thread while another is trying to write to it) the obvious
course is to wrap the greeting in the MySession's lock. The trouble is then
knowing when the write("greeting") has finished to release the lock -
await() throws the IllegalStateException ("in I/O thread"...) as one might
expect, but using a future (same as in messageReceived(...) below) throws an
exception because it's trying to release a lock in a different thread from
the one which held it.

What do I need to do to handle in-progress greetings gracefully at shutdown?
Is there a better way of handling the shutdown/session management overall?
(ExecutorUtil.terminate() doesn't wait for in-progress processing to
finish.)


Below is a simplified and abridged version of my code (missing exception
handling etc.) - at the very least please have a look at
channelConnected(...):


Thanks for your help!




private void main() {
    ExecutorService boss = Executors.newCachedThreadPool();
    ExecutorService workers = Executors.newCachedThreadPool();

    final ServerBootstrap bootstrap = new ServerBootstrap(new
NioServerSocketChannelFactory(boss, workers));

    bootstrap.setPipelineFactory(myPipelineFactory);
    final Channel bind = bootstrap.bind(new
InetSocketAddress(myPortNumber));

    Runtime.getRuntime().addShutdownHook(new Thread() {
        public void run() {
            // synchronously close sessions and their parent channels
            myChannelHandler.shutDown();

            // once the channels have closed, the rest of the server will
shut itself down
            bind.close().awaitUninterruptibly(10000);
            bootstrap.releaseExternalResources();
        }
    });
}




// In myChannelHandler:

public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
throws Exception {
        String requestRawFromSocket = (String) e.getMessage();
        final mySession session = (MySession)ctx.getAttachment();

        if (session.getLock().tryLock()) {
            String reply =
requestHandler.handleRequest(requestRawFromSocket, session);

            ChannelFuture future = ctx.getChannel().write(reply);

            // make sure we wait for the reply to be sent before we release
lock.
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture)
throws Exception {
                    session.getLock().unlock();
                }
            });
            session.getLock().unlock();
        }
}

public synchronized void shutDown() {
    for (MySession session : allSessions)
        session.markForTermination();

    while (mySessions.size() > 0)
        terminateMarkedSessions();
}

// (This is also routinely called by a repeating scheduled task for
asynchronous cleanup of 
//  sessions after e.g. a logout command)

public synchronized void terminateMarkedSessions() {
    Queue<MySession> sessionsToKill = new LinkedList<MySession>();

    // decide which sessions need to be killed
    for (MySession session : allSessions) 
        if (session.isMarkedForTermination() && session.getLock().tryLock()) 
            sessionsToKill.add(session);

    for (MySession session : sessionsToKill) {
        session.getChannel().close().awaitUninterruptibly();
        mySessions.remove(session);
    }
}

@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
    channels.add(ctx.getChannel());
    final MySession session = new MySession(ctx.getChannel());
    ctx.setAttachment(session);
    mySessions.addSession(session);
    
    // what happens if server shuts down while executing this?
    ctx.getChannel().write("This is a greeting message");
}



--
View this message in context: http://netty-forums-and-mailing-lists.685743.n2.nabble.com/Graceful-shutdown-tp6495032p6495032.html
Sent from the Netty User Group mailing list archive at Nabble.com.