Servidor NIO TCP de alta carga Java

high-load

Question

Como parte de mi investigación, estoy escribiendo un servidor de eco TCP / IP de alta carga en Java. Quiero atender alrededor de 3-4k de clientes y ver los mensajes máximos posibles por segundo que puedo exprimir. El tamaño del mensaje es bastante pequeño, hasta 100 bytes. Este trabajo no tiene ningún propósito práctico, solo una investigación.

De acuerdo con las numerosas presentaciones que he visto (puntos de referencia de HornetQ, charlas de LMAX Disruptor, etc.), los sistemas reales de carga alta tienden a atender millones de transacciones por segundo (creo que Disruptor mencionó aproximadamente 6 mils y Hornet - 8.5). Por ejemplo, esta publicación indica que es posible alcanzar hasta 40M MPS. Así que lo tomé como una estimación aproximada de lo que debería ser capaz de hacer el hardware moderno.

Escribí el servidor NIO de un solo hilo más simple y lancé una prueba de carga. Me sorprendió un poco que pudiera obtener solo unos 100k MPS en localhost y 25k con redes reales. Los números parecen bastante pequeños. Estaba probando en Win7 x64, core i7. En cuanto a la carga de la CPU, solo un núcleo está ocupado (lo que se espera en una aplicación de un solo hilo), mientras que el resto permanece inactivo. Sin embargo, aunque cargue los 8 núcleos (incluido el virtual) no tendré más de 800k MPS, ni siquiera cerca de 40 millones :)

Mi pregunta es: ¿cuál es un patrón típico para entregar cantidades masivas de mensajes a los clientes? ¿Debo distribuir la carga de red a través de varios sockets diferentes dentro de una única JVM y usar algún tipo de equilibrador de carga como HAProxy para distribuir la carga a múltiples núcleos? ¿O debería mirar hacia el uso de varios selectores en mi código NIO? ¿O tal vez incluso distribuir la carga entre múltiples JVM y usar Chronicle para construir una comunicación entre procesos? ¿Las pruebas en un sistema operativo del lado del servidor como CentOS harán una gran diferencia (tal vez sea Windows lo que ralentiza las cosas)?

A continuación se muestra el código de ejemplo de mi servidor. Siempre responde con "ok" a cualquier dato entrante. Sé que en el mundo real necesitaría hacer un seguimiento del tamaño del mensaje y estar preparado para que un solo mensaje se divida entre varias lecturas, pero me gustaría mantener las cosas súper simples por ahora.

public class EchoServer {

private static final int BUFFER_SIZE = 1024;
private final static int DEFAULT_PORT = 9090;

// The buffer into which we''ll read data when it''s available
private ByteBuffer readBuffer = ByteBuffer.allocate(BUFFER_SIZE);

private InetAddress hostAddress = null;

private int port;
private Selector selector;

private long loopTime;
private long numMessages = 0;

public EchoServer() throws IOException {
    this(DEFAULT_PORT);
}

public EchoServer(int port) throws IOException {
    this.port = port;
    selector = initSelector();
    loop();
}

private void loop() {
    while (true) {
        try{
            selector.select();
            Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selectedKeys.remove();

                if (!key.isValid()) {
                    continue;
                }

                // Check what event is available and deal with it
                if (key.isAcceptable()) {
                    accept(key);
                } else if (key.isReadable()) {
                    read(key);
                } else if (key.isWritable()) {
                    write(key);
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
    }
}

private void accept(SelectionKey key) throws IOException {
    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();

    SocketChannel socketChannel = serverSocketChannel.accept();
    socketChannel.configureBlocking(false);
    socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
    socketChannel.setOption(StandardSocketOptions.TCP_NODELAY, true);
    socketChannel.register(selector, SelectionKey.OP_READ);

    System.out.println("Client is connected");
}

private void read(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();

    // Clear out our read buffer so it''s ready for new data
    readBuffer.clear();

    // Attempt to read off the channel
    int numRead;
    try {
        numRead = socketChannel.read(readBuffer);
    } catch (IOException e) {
        key.cancel();
        socketChannel.close();

        System.out.println("Forceful shutdown");
        return;
    }

    if (numRead == -1) {
        System.out.println("Graceful shutdown");
        key.channel().close();
        key.cancel();

        return;
    }

    socketChannel.register(selector, SelectionKey.OP_WRITE);

    numMessages++;
    if (numMessages%100000 == 0) {
        long elapsed = System.currentTimeMillis() - loopTime;
        loopTime = System.currentTimeMillis();
        System.out.println(elapsed);
    }
}

private void write(SelectionKey key) throws IOException {
    SocketChannel socketChannel = (SocketChannel) key.channel();
    ByteBuffer dummyResponse = ByteBuffer.wrap("ok".getBytes("UTF-8"));

    socketChannel.write(dummyResponse);
    if (dummyResponse.remaining() > 0) {
        System.err.print("Filled UP");
    }

    key.interestOps(SelectionKey.OP_READ);
}

private Selector initSelector() throws IOException {
    Selector socketSelector = SelectorProvider.provider().openSelector();

    ServerSocketChannel serverChannel = ServerSocketChannel.open();
    serverChannel.configureBlocking(false);

    InetSocketAddress isa = new InetSocketAddress(hostAddress, port);
    serverChannel.socket().bind(isa);
    serverChannel.register(socketSelector, SelectionKey.OP_ACCEPT);
    return socketSelector;
}

public static void main(String[] args) throws IOException {
    System.out.println("Starting echo server");
    new EchoServer();
}
}

Answer 1

Alcanzarás unos cientos de miles de solicitudes por segundo con hardware normal. Al menos esa es mi experiencia al tratar de crear soluciones similares, y el punto de referencia de los marcos web de Tech Empower también parece estar de acuerdo.

El mejor enfoque, generalmente, depende de si tiene cargas enlazadas a io o cpu.

Para las cargas ligadas a io (latencia alta), debe hacer async io con muchos subprocesos. Para obtener el mejor rendimiento, debe intentar anular las transferencias entre hilos lo más posible. Por lo tanto, tener un subproceso de selección dedicado y otro conjunto de subprocesos para el procesamiento es más lento que tener un subproceso donde cada subproceso realiza una selección o un procesamiento, de modo que una solicitud es manejada por un solo subproceso en el mejor de los casos (si io está disponible de inmediato). Este tipo de configuración es más complicada de codificar pero rápida, y no creo que ningún marco web asíncrono explote esto completamente.

Para las cargas vinculadas a cpu, un hilo por solicitud suele ser el más rápido, ya que evita los cambios de contexto.

Answer 2

Tu lógica alrededor de la escritura es defectuosa. Debes intentar la escritura inmediatamente tienes datos para escribir. Si write() devuelve cero, entonces es el momento de registrarse para OP_WRITE, vuelva a intentar la escritura cuando el canal se OP_WRITE escribir, y cancele el registro de OP_WRITE cuando la escritura haya tenido éxito. Estás agregando una cantidad masiva de latencia aquí. Usted está agregando aún más latencia al cancelar el registro de OP_READ mientras está haciendo todo eso.

Answer 3

what is a typical pattern for serving massive amounts of messages to clients?

Hay muchos patrones posibles: Una forma fácil de utilizar todos los núcleos sin pasar por múltiples JVMS es:

1. Haga que un solo hilo acepte conexiones y lea utilizando un selector.
2. Una vez que tenga suficientes bytes para constituir un solo mensaje, páselo a otro núcleo utilizando una construcción como un búfer de anillo. El framework Java de Disruptor es una buena combinación para esto. Este es un buen patrón si el procesamiento necesario para saber qué es un mensaje completo es liviano. Por ejemplo, si tiene un protocolo prefijado de longitud, podría esperar hasta obtener el número esperado de bytes y luego enviarlo a otro hilo. Si el análisis del protocolo es muy intenso, puede abrumar este único hilo que le impide aceptar conexiones o leer bytes de la red.
3. En los subprocesos de trabajo, que reciben datos de un búfer de anillo, realice el procesamiento real.
4. Usted escribe las respuestas en sus subprocesos de trabajo o a través de otro subproceso de agregador.

Eso es lo esencial. Hay muchas más posibilidades aquí y la respuesta realmente depende del tipo de aplicación que está escribiendo. Algunos ejemplos son:

1. Una aplicación CPU sin estado pesado dice una aplicación de procesamiento de imágenes. La cantidad de trabajo de CPU / GPU realizado por solicitud probablemente será significativamente mayor que la sobrecarga generada por una solución de comunicación entre subprocesos muy ingenua. En este caso, una solución fácil es un conjunto de subprocesos de trabajo que extraen el trabajo de una sola cola. Observe cómo se trata de una cola única en lugar de una cola por trabajador. La ventaja es que esto es inherentemente equilibrado de carga. Cada trabajador termina su trabajo y luego simplemente sondea la cola de múltiples consumidores de un solo productor. Aunque esta es una fuente de controversia, el trabajo de procesamiento de imágenes (¿segundos?) Debería ser mucho más costoso que cualquier alternativa de sincronización.
2. Una aplicación de E / S pura, por ejemplo, un servidor de estadísticas que solo incrementa algunos contadores para una solicitud: aquí no hace casi ningún trabajo pesado de CPU. La mayor parte del trabajo es solo leer bytes y escribir bytes. Es posible que una aplicación de subprocesos múltiples no le proporcione beneficios significativos aquí De hecho, incluso podría ralentizar las cosas si el tiempo que lleva poner en cola los elementos es más que el tiempo necesario para procesarlos. Un solo servidor Java de subprocesos debería poder saturar un enlace 1G fácilmente.

3. Aplicaciones con estado que requieren cantidades moderadas de procesamiento, por ejemplo, una aplicación comercial típica: aquí, cada cliente tiene algún estado que determina cómo se maneja cada solicitud. Suponiendo que vayamos con múltiples subprocesos ya que el procesamiento no es trivial, podríamos afinizar a los clientes con ciertos subprocesos. Esta es una variante de la arquitectura del actor:

   i) Cuando un cliente primero conecta el hash con un trabajador. Es posible que desee hacer esto con algún ID de cliente, de modo que si se desconecta y se vuelve a conectar todavía se asigne al mismo trabajador / actor.

   ii) Cuando el hilo del lector lee una solicitud completa, colóquelo en el buffer de anillo para el trabajador / actor adecuado. Dado que el mismo trabajador siempre procesa un cliente particular, todo el estado debe ser subproceso local, haciendo que toda la lógica de procesamiento sea simple y de un solo subproceso.

   iii) El subproceso de trabajo puede escribir solicitudes. Siempre intente simplemente hacer un write (). Si no se pudieron escribir todos sus datos solo entonces se registra en OP_WRITE. El subproceso de trabajo solo necesita realizar llamadas de selección si en realidad hay algo pendiente La mayoría de las escrituras deberían tener éxito haciendo esto innecesario. El truco aquí es equilibrar las llamadas seleccionadas y sondear el búfer del anillo para obtener más solicitudes. También puede emplear un solo hilo de escritor cuya única responsabilidad es escribir solicitudes. Cada subproceso de trabajo puede poner sus respuestas en un búfer de anillo conectándolo a este único subproceso de escritura. El único hilo del escritor round-robin sondea cada búfer de anillo entrante y escribe los datos a los clientes. De nuevo, se aplica la advertencia de intentar escribir antes de seleccionar, al igual que el truco sobre el equilibrio entre varios buffers de llamada y llamadas selectas.

Como señala usted hay muchas otras opciones:

Should I distribute networking load over several different sockets inside a single JVM and use some sort of load balancer like HAProxy to distribute load to multiple cores?

Puede hacer esto, pero en mi humilde opinión no es el mejor uso para un equilibrador de carga. Esto le permite comprar JVM independientes que pueden fallar por sí solas, pero probablemente serán más lentas que escribir una aplicación JVM única que tenga múltiples subprocesos. Sin embargo, la aplicación en sí puede ser más fácil de escribir, ya que será de un solo hilo.

Or I should look towards using multiple Selectors in my NIO code?

Usted puede hacer esto también. Mire la arquitectura de Ngnix para obtener algunos consejos sobre cómo hacer esto.

Or maybe even distribute the load between multiple JVMs and use Chronicle to build an inter-process communication between them? Esta es también una opción. Chronicle le da la ventaja de que los archivos asignados en memoria son más resistentes a un proceso que se cierra en el medio. Aún obtienes bastante rendimiento ya que toda la comunicación se realiza a través de la memoria compartida.

Will testing on a proper serverside OS like CentOS make a big difference (maybe it is Windows that slows things down)?

No sé sobre esto. Improbable. Si Java usa las API nativas de Windows al máximo, no debería importar tanto. Tengo muchas dudas sobre la cifra de 40 millones de transacciones / seg (sin una pila de redes de espacio de usuario + UDP) pero las arquitecturas que enumeré deberían funcionar bastante bien.

Estas arquitecturas tienden a funcionar bien, ya que son arquitecturas de un solo escritor que utilizan estructuras de datos basadas en arreglos limitados para la comunicación entre subprocesos. Determine si el multihilo es la respuesta correcta. En muchos casos no es necesario y puede llevar a una desaceleración.

Otra área a considerar son los esquemas de asignación de memoria. Específicamente, la estrategia para asignar y reutilizar los buffers podría generar beneficios significativos. La estrategia correcta de reutilización del búfer depende de la aplicación. Mire esquemas como la asignación de memoria de amigos, la asignación de arenas, etc. para ver si pueden beneficiarlo. El JVM GC funciona bastante bien para la mayoría de las cargas de trabajo, por lo que siempre debe medir antes de tomar esta ruta.

El diseño del protocolo también tiene un gran efecto en el rendimiento. Tiendo a preferir los protocolos prefijados de longitud porque te permiten asignar buffers del tamaño correcto, evitando listas de buffers y / o la fusión de buffers. Los protocolos prefijados de longitud también facilitan la decisión de cuándo entregar una solicitud, simplemente marque el num bytes == expected . El análisis real se puede hacer por el hilo de los trabajadores. La serialización y la deserialización se extienden más allá de los protocolos con prefijo de longitud. Los patrones como los patrones de peso mosca sobre buffers en lugar de asignaciones ayudan aquí. Mira a SBE para algunos de estos principios.

Como puedes imaginar, un tratado completo podría escribirse aquí. Esto debería ponerte en la dirección correcta. Advertencia: siempre mida y asegúrese de que necesita más rendimiento que la opción más simple. Es fácil dejarse atrapar por un agujero negro sin fin de mejoras de rendimiento.