Application level load balancer for pool of processes

Question

We have legacy monolithic software in C++ that acts like a request-reply TCP server. This software is single-threaded and may process just one request simultaneously. Currently we have fixed pool of such processes to serve multiple clients in parallel.

Periodically clients experience significant delays in request processing due to the large amount of messages. Currently we have an idea to solve this problem by introducing kind of Proxy between clients and workers:

We want the following functionality for this Proxy:

1. Application-level load balancing: spreading requests between workers by checking request context and client id
2. Control and monitor life cycle of the worker processes
3. Spawn additional worker processes (on different PCs) to handle peaks

In fact we want it to behave like an ExecutorService in Java, but with worker processes instead of threads. Current idea is to implement this Balancer in Java based on Jetty or Tomcat server, with internal message queue and servlets forwarding requests to the worker processes.

But I'm wondering: is there existing solutions (preferable in Java) that would automate this process? What would be the easiest way to implement such a Proxy?

UPDATE:

What I do with request context - well, that C++ server is really messy software. In fact every time it receives different context it updates internal cache accordingly to match that context. For example if you request that server to give you some data in English, then it reloads the internal cache to English. If next request is in French, then it reloads the cache again. Obviously I would like to minimize number of cache reloads by forwarding requests a bit more intelligently.

The communication protocol is homemade (based on TCP/IP), but it's relatively easy to extract context part from it.

Currently load balancing is implemented on the client side, so each client is configured to know all the server nodes and sends requests to them in round robin fashion. There are several problems of this approach: complicated connection management on the client side, incorrect work with multiple clients who don't know about each other, can't manage node life cycle.. We cannot solve the listed problems with refactoring.

Most likely we will end up with homemade solution for forwarding, but I'm still wondering if there are existing products at least for process management?? Ideally this would be Java application server that can:

• Spawn children nodes (another Java processes)
• Monitor their lifecycle
• Communicate with them by some protocol

Maybe this functionality is already implemented in some existing application servers? This would greatly simplify the problem!

Answer 1

Regarding process management you can easily achieve your goal by mixing functionalities of Apache Commons Exec library which can help spawning new instances of workers with Apache Commons Pool library which will manage running instances.

The implementation is really easy because commons pool will ensure that you can use one object at the time until it is returned to the pool. If object is not returned to the pool, commons pool will spawn new instance for you. You can controll lifetime of your workers either by adding watchdog service (from apache commons exec) - watchdog can kill an instance unused for some time or you can also use commons pool itself for example by calling pool.clearOldest(). You can also see how many requests are processed (how many workers are active) at the moment by calling pool.getNumActive(). Reffer to javadoc of GenericKeyedObjectPool to see more.

The implementation can be done with one simple servlet running on Tomcat server. This servlet will instantiate the pool and simply ask the pool for new worker by calling pool.borowObject(parameters). Inside parameters you define what characteristics should have your worker to handle request (in your case parameters should include the language). In case no such worker is available (for example no worker for french langauge) pool will spawn new worker for you. Also if there is a worker but the worker is currently handling another request, pool will also spawn a new worker for you (so you will have two workers handling the same language). Worker will be back ready to handle new request when you call pool.returnObject(parameters, instance).

The whole implementation took me less than 200 lines of code (see below for complete code). The code includes a situation when a worker process gets killed from outside or will crash (see WorkersFactory.activateObject()).

IMHO: using Apache Cammel is not good option for you, because it too big tool and it is designed as a mediation bus between different messages format. You don't need to do transformations, you don't need to handle different formats of messages. Go for the simple solution.

package com.myapp;

import org.apache.commons.exec.CommandLine;
import org.apache.commons.exec.DefaultExecutor;
import org.apache.commons.exec.ExecuteWatchdog;
import org.apache.commons.pool2.BaseKeyedPooledObjectFactory;
import org.apache.commons.pool2.KeyedPooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Objects;

public class BalancingServlet extends javax.servlet.http.HttpServlet {

    private final WorkersPool workersPool;

    public BalancingServlet() {
        workersPool = new WorkersPool(new WorkersFactory());
    }


    protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {

    }

    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        response.getWriter().println("Balancing");

        String language = request.getParameter("language");
        String someOtherParam = request.getParameter("other");
        WorkerParameters workerParameters = new WorkerParameters(language, someOtherParam);

        String requestSpecificParam1 = request.getParameter("requestParam1");
        String requestSpecificParam2 = request.getParameter("requestParam2");

        try {
            WorkerInstance workerInstance = workersPool.borrowObject(workerParameters);
            workerInstance.handleRequest(requestSpecificParam1, requestSpecificParam2);
            workersPool.returnObject(workerParameters, workerInstance);

        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}

class WorkerParameters {
    private final String workerLangauge;
    private final String someOtherParam;

    WorkerParameters(String workerLangauge, String someOtherParam) {
        this.workerLangauge = workerLangauge;
        this.someOtherParam = someOtherParam;
    }

    public String getWorkerLangauge() {
        return workerLangauge;
    }

    public String getSomeOtherParam() {
        return someOtherParam;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        WorkerParameters that = (WorkerParameters) o;

        return Objects.equals(this.workerLangauge, that.workerLangauge) && Objects.equals(this.someOtherParam, that.someOtherParam);
    }

    @Override
    public int hashCode() {
        return Objects.hash(workerLangauge, someOtherParam);
    }
}

class WorkerInstance {
    private final Thread thread;
    private WorkerParameters workerParameters;

    public WorkerInstance(final WorkerParameters workerParameters) {
        this.workerParameters = workerParameters;

        // launch the process here   
        System.out.println("Spawing worker for language: " + workerParameters.getWorkerLangauge());

        // use commons Exec to spawn your process using command line here

        // something like


        thread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    String line = "C:/Windows/notepad.exe" ;
                    final CommandLine cmdLine = CommandLine.parse(line);

                    final DefaultExecutor executor = new DefaultExecutor();
                    executor.setExitValue(0);
//                    ExecuteWatchdog watchdog = new ExecuteWatchdog(60000); // if you want to kill process running too long
//                    executor.setWatchdog(watchdog);

                    int exitValue = executor.execute(cmdLine);
                    System.out.println("process finished with exit code: " + exitValue);
                } catch (IOException e) {
                    throw new RuntimeException("Problem while executing application for language: " + workerParameters.getWorkerLangauge(), e);
                }


            }
        });

        thread.start();


        System.out.println("Process spawned for language: " + workerParameters.getWorkerLangauge());


    }

    public void handleRequest(String someRequestParam1, String someRequestParam2) {
        System.out.println("Handling request for extra params: " + someRequestParam1 + ", " + someRequestParam2);

        // communicate with your application using parameters here

        // communcate via tcp or whatever protovol you want using extra parameters: someRequestParam1, someRequestParam2


    }

    public boolean isRunning() {
        return thread.isAlive();
    }


}

class WorkersFactory extends BaseKeyedPooledObjectFactory<WorkerParameters, WorkerInstance> {

    @Override
    public WorkerInstance create(WorkerParameters parameters) throws Exception {
        return new WorkerInstance(parameters);
    }

    @Override
    public PooledObject<WorkerInstance> wrap(WorkerInstance worker) {
        return new DefaultPooledObject<WorkerInstance>(worker);
    }

    @Override
    public void activateObject(WorkerParameters worker, PooledObject<WorkerInstance> p)
            throws Exception {
        System.out.println("Activating worker for lang: " + worker.getWorkerLangauge());

        if  (! p.getObject().isRunning()) {
            System.out.println("Worker for lang: " + worker.getWorkerLangauge() + " stopped working, needs to respawn it");
            throw new RuntimeException("Worker for lang: " + worker.getWorkerLangauge() + " stopped working, needs to respawn it");
        }
    }

    @Override
    public void passivateObject(WorkerParameters worker, PooledObject<WorkerInstance> p)
            throws Exception {
        System.out.println("Passivating worker for lang: " + worker.getWorkerLangauge());
    }

}

class WorkersPool extends GenericKeyedObjectPool<WorkerParameters, WorkerInstance> {

    public WorkersPool(KeyedPooledObjectFactory<WorkerParameters, WorkerInstance> factory) {
        super(factory);
    }
}

Answer 2

It looks like you are looking for a messaging system. Apache Camel has a lot of compoonents to integrate different protocols and add custom processing logic (using XML or java API). Apache Camel has implemented a lot of the (Enterprise Integration Patterns).

It has an integration with Apache MINA which can also be a good starting point to look at.

It's not clear how you will be able to start new instances on other computers on the fly. I think you will at least need some agents running on these machines which you can request to launch a new server.