How to access memcached asynchronously in netty

Question

I am writing a server in netty where I need to call memcached. I use spymemcached and can easily make a synchronous memcached call. I would like this memcached call to be asynchronous. Is it possible? The examples provided by netty do not seem useful.

I tried using callbacks: I created the ExecutorService pool in my handler and sent a callback handler to this pool. Like this:

public class MyHandler extends ChannelInboundMessageHandlerAdapter <MyPOJO> implements CallbackInterface {

  ... private static ExecutorService pool = Executors.newFixedThreadPool(20); @Override public void messageReceived(ChannelHandlerContext ctx, MyPOJO pojo) { ... CallingbackWorker worker = new CallingbackWorker(key, this); pool.submit(worker); ... } public void myCallback() { //get response this.ctx.nextOutboundMessageBuf().add(response); } 

}

CallingbackWorker as follows:

public class CallingbackWorker implements Callable {

  public CallingbackWorker(String key, CallbackInterface c) { this.c = c; this.key = key; } public Object call() { //get value from key c.myCallback(value); } 

However, when I do this, this.ctx.nextOutboundMessageBuf() in myCallback gets stuck.

So, overall, my question is: how to make asynchronous memcached calls in Netty?

Answer 1

There are two problems here: a small problem with the way you are trying to encode it, and a larger one, with many libraries that provide async service calls but have no way to fully use them in async, like Netty. This forces users into suboptimal hacks like this, or less bad, but still not ideal approaches, I can in a moment.

The coding problem first. The problem is that you are trying to call the ChannelHandlerContext method from a thread other than the one associated with your handler, which is not allowed. This is pretty easy to fix as shown below. You can encode it in several other ways, but this is probably the easiest:

 private static ExecutorService pool = Executors.newFixedThreadPool(20); public void channelRead(final ChannelHandlerContext ctx, final Object msg) { //... final GetFuture<String> future = memcachedClient().getAsync("foo", stringTranscoder()); // first wait for the response on a pool thread pool.execute(new Runnable() { public void run() { String value; Exception err; try { value = future.get(3, TimeUnit.SECONDS); // or whatever timeout you want err = null; } catch (Exception e) { err = e; value = null; } // put results into final variables; compiler won't let us do it directly above final fValue = value; final fErr = err; // now process the result on the ChannelHandler thread ctx.executor().execute(new Runnable() { public void run() { handleResult(fValue, fErr); } }); } }); // note that we drop through to here right after calling pool.execute() and // return, freeing up the handler thread while we wait on the pool thread. } private void handleResult(String value, Exception err) { // handle it } 

This will work, and may be enough for your application. But you have a fixed-size thread pool, so if you ever handle so many more than 20 concurrent connections, this will become a bottleneck. You can increase the pool size or use unlimited, but at this point you can also work under Tomcat, as memory consumption and context overhead begin to become problems, and you lose the scalability, which was attractive to Netty in the first place!

And the whole point is that Spymemcached is based on NIO, is event-driven and uses only one thread for all its work, but does not make it possible to fully use its event-driven nature. I expect them to fix this for too long, just as Netty 4 and Cassandra recently provided callback methods (listeners) for Future objects.

Meanwhile, being in the same boat as you, I explored alternatives and not too happy with what I found, I wrote (yesterday) a Future tracker class , which can interrogate up to a thousand futures at a custom rate and return you to the stream ( Artist) of your choice when they are completed. Only one thread is used for this. I put it on GitHub if you want to try it, but be careful as they are still wet. I experienced this a lot last day and even with 10,000 concurrent scammers. Future objects, polling once a millisecond, its CPU usage is negligible, although it starts to grow above 10,000. Using it, the above example looks like this

 // in some globally-accessible class: public static final ForeignFutureTracker FFT = new ForeignFutureTracker(1, TimeUnit.MILLISECONDS); // in a handler class: public void channelRead(final ChannelHandlerContext ctx, final Object msg) { // ... final GetFuture<String> future = memcachedClient().getAsync("foo", stringTranscoder()); // add a listener for the Future, with a timeout in 2 seconds, and pass // the Executor for the current context so the callback will run // on the same thread. Global.FFT.addListener(future, 2, TimeUnit.SECONDS, ctx.executor(), new ForeignFutureListener<String,GetFuture<String>>() { public void operationSuccess(String value) { // do something ... ctx.fireChannelRead(someval); } public void operationTimeout(GetFuture<String> f) { // do something ... } public void operationFailure(Exception e) { // do something ... } }); } 

You do not want more than one or two instances of the FFT to be involved at any time, or they could become a processor leak. But one instance can handle thousands of outstanding futures; the only reason you need to have a second is to handle calls with a higher delay, such as S3, at a lower polling rate, say, 10-20 milliseconds.

One of the drawbacks of the survey approach is that it adds a small amount of latency. For example, polling once a millisecond, on average it will add 500 μs to the response time. This will not be a problem for most applications, and I think that this is more than offset by the savings in memory and CPU compared to the thread pool.

I expect that in a year or so this will be a problem without problems, as more asynchronous clients provide callback mechanisms, allowing full use of the NIO and event-driven model.