Reimplementing RMI

Today, on our forums, a user asked how to implement an RMI like solution:

My clustered app will be a client of an external service (via a socket connection) where only ONE node in the cluster will maintain the connection. All the other nodes will need to utilize the external service via the node connected to it.

An interesting puzzle. It's basically RMI. Should one reimplement RMI with Terracotta? That wasn't the point -- I can't resist a challenge like this, so I whipped up a solution. Ultimately, it ended up being a combination of a queue to send the method invocation, and some wait/notify to wait for the response.

As I was chatting about the solution with DSO Guy Steve Harris I remarked that I often come across questions of the sort "how do I do this in Terracotta?" I said that the answer was remarkably easy - though I think it just doesn't occur to people how easy it really is.

The great thing about Terracotta is that it really gives you just Java. By extending the Java Memory Model, Terracotta makes sure that memory and synchronization work exactly the same as in a single JVM. So the answer to the question "how do I do this in Terracotta?" is extremely easy: How do you do this in a single JVM with multiple threads? Whatever your answer is, that's the answer to doing it with Terracotta.

To illustrate this point, let's see how I solved the RMI question. Basically, I implemented a Dynamic Proxy (with some help from Hung on writing a Dynamic Proxy here which neatly wraps up the guts of the implementation so it can be wrapped around any implementation.

As a result of using Dynamic Proxy, we already get a Method and Arguments that we have to handle, so to execute this method on a specific node is the same as executing it on a separate thread. To do that, use a queue. And to signal the response, use wait and notify.

Here's the code:

import java.util.concurrent.*;
import java.lang.reflect.*;

public class RemoteInvoker implements InvocationHandler, Runnable
{
    private BlockingQueue<MethodArguments> queue = new LinkedBlockingQueue<MethodArguments>();

    private final Object instance;

    public RemoteInvoker(Object instance)
    {
        this.instance = instance;
        start();
    }

    private static class MethodArguments
    {
        public final Object proxy;
        public final Method method;
        public final Object[] args;
        private MethodResult result;

        public MethodArguments(Object proxy, Method method, Object[] args)
        {
            this.proxy = proxy;
            this.method = method;
            this.args = args;
        }

        public synchronized MethodResult getResult() throws InterruptedException
        {
           while (result == null) { wait(); }
           return result;
        }

        public synchronized void setResult(MethodResult result)
        {
            this.result = result;
            notify();
        }
    }

    private static class MethodResult
    {
        public final Object object;
        public final Exception exception;

        public MethodResult(Object object, Exception exception)
        {
            this.object = object;
            this.exception = exception;
        }
    }

    private void start()
    {
        Thread t = new Thread(this);
//        t.setDaemon(true);
        t.start();
        t.start();
    }

    public void run()
    {
        synchronized (instance) {
            System.out.println("I am servicing requests...");

            MethodArguments arguments;
            while (true) {
                try {
                    arguments = queue.take();
                    try {
                        Object value = arguments.method.invoke(instance, arguments.args);
                        arguments.setResult(new MethodResult(value, null));
                    } catch (Exception e) {
                        arguments.setResult(new MethodResult(null, e));
                    }
                } catch (InterruptedException e) {
                    // do nothing
                }
            }
        }
    }

    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable
    {
        MethodArguments arguments = new MethodArguments(proxy, method, args);
        queue.put(arguments);
        MethodResult result;

        result = arguments.getResult();
        if (result.exception != null) {
            throw result.exception;
        }

        return result.object;
    }

And here's how to use it:

import java.lang.reflect.*;

public class Test implements TestInterface
{
    int counter = 0;

    public int count(int count)
    {
        System.out.println("Incrementing counter by: " + count);

        counter += count;
        return counter;
    }

    public static void main(String[] args)
    {
        Test t = new Test();
        RemoteInvoker invoker = new RemoteInvoker(t);
        TestInterface proxy = (TestInterface) Proxy.newProxyInstance(t.getClass().getClassLoader(), new Class[] { TestInterface.class } , invoker);
        System.out.println("Proxy returns: " + proxy.count(3));
    }
}

What I really like about this solution, and Terracotta in general, is that it works exactly the same in a single JVM as many JVMs. The key difference to many JVMs is that by clustering the Queue, the execution is transferred across physical JVMs, and the wait/notify sends the response back - no different than cross-thread communication in a single JVM.

For the complete instructions on running this with Terracotta, read the whole thread.

Btw, if you read the code, maybe you were wondering why run() method is synchronized against instance. Well, getting back to the request from the user, the original requirement was that only one instance can be "connected" - in other words servicing requests. The synchronized makes sure only one instance is servicing the queue - whether it's a single JVM or many JVMs.