java - tuning - ¿Por qué veo bajadas de rendimiento en el arranque del servidor Netty aceptando nuevos canales?

netty server

Question

Todavía estoy usando Netty 3.10. Escribí una prueba de unidad para verificar el rendimiento del hilo del jefe de Netty. Utilizo un bootstrap simple del servidor Netty dentro del hilo principal de la prueba de unidad y genero 100 clientes Java Sync-IO dentro de un grupo de subprocesos en caché. Me di cuenta de las caídas de rendimiento que creo que son extrañas. Cada cliente abre un socket, escribe datos y se cierra, registrando la duración (ms) después de cerrar. Mi prueba unitaria está adjunta. La salida típica de mi prueba unitaria es, en orden dado:

1. 43 x Cliente hecho. Duración: 0
2. 26 x Cliente hecho. Duración: 16
3. 16 x Cliente hecho. Duración: 0
4. 3 x Cliente hecho. Duración: 517
5. 11 x Cliente hecho. Duración: 3003
6. 1 x Cliente hecho. Duración: 6036

Entonces, hubo 1 cliente que tuvo que esperar 6 segundos para obtener un canal TCP / IP abierto y 11 clientes que tuvieron que esperar 3 segundos. También verifiqué dónde se pierde / pierde el tiempo. Siempre es new Socket(String,int) en el lado del cliente. En el lado del servidor, el tiempo ya no existe cuando se activa la fábrica de tuberías.

¿El modelo de enhebrado de mi unidad es la causa de este o realmente Netty bootstrap / boss?

import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.junit.After;
import org.junit.Assert;
import org.junit.Test;

public class NettyServerBossTest {

    private static final String SRV_HOST = "localhost";
    private static final int SRV_PORT = 8080;
    private static final byte[] MSG = "Hello world!".getBytes(Charset.forName("UTF-8"));
    private static final int WAIT_MAX_MILLIS = 10 * 1000;

    private final ChannelGroup channels = new DefaultChannelGroup();

    private final int expected = 100;
    private final AtomicInteger actual = new AtomicInteger();

    private volatile boolean failed;
    private ExecutorService clientThreads;
    private Throwable cause;
    private ServerBootstrap bootstrap;

    @Test
    public void test() {
        createServer();
        createClients();
        awaitClients();
        verifyFailure();
    }

    private void awaitClients() {
        final long startMillis = System.currentTimeMillis();
        final long maxMillis = startMillis + WAIT_MAX_MILLIS;
        while ((this.actual.get() < this.expected) && !isFailed() && (System.currentTimeMillis() < maxMillis)) {
            try {
                Thread.sleep(250L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("Total duration: " + (System.currentTimeMillis() - startMillis));
        Assert.assertEquals(this.expected, this.actual.get());
    }

    private void createClients() {
        this.clientThreads = Executors.newCachedThreadPool();
        for (int i = 0; i < this.expected; i++) {
            this.clientThreads.execute(new PlainSocketClient());
        }
    }

    private void closeChannels() {
        try {
            this.channels.close().await(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void createServer() {
        final ExecutorService bosses = Executors.newCachedThreadPool();
        final ExecutorService workers = Executors.newCachedThreadPool();
        final ChannelFactory factory = new NioServerSocketChannelFactory(bosses, workers);
        this.bootstrap = new ServerBootstrap(factory);
        this.bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() {
                return Channels.pipeline(new DiscardServerHandler());
            }
        });
        this.bootstrap.setOption("child.tcpNoDelay", Boolean.TRUE);
        this.bootstrap.setOption("child.keepAlive", Boolean.TRUE);
        this.bootstrap.bind(new InetSocketAddress(SRV_HOST, SRV_PORT));
    }

    /**
     * Fail unit test
     *
     * @param cause
     *            cause of failure
     */
    public synchronized void setCause(Throwable cause) {
        if (!this.failed && (cause == null)) {
            this.failed = true;
            this.cause = cause;
        }
    }

    @After
    public void after() {
        closeChannels();
        if (this.clientThreads != null) {
            this.clientThreads.shutdownNow();
        }
        if (this.bootstrap != null) {
            this.bootstrap.releaseExternalResources();
        }
    }

    /**
     * Check if unit test has failed
     *
     * @return <code>true</code> if failed, <code>false</code> if still OK
     */
    public boolean isFailed() {
        return this.failed;
    }

    /**
     * Get cause of failure
     *
     * @return cause or <code>null</code>
     */
    public synchronized Throwable getCause() {
        return this.cause;
    }

    /**
     * Make sure test has not failed with exception
     */
    public void verifyFailure() {
        if (this.failed) {
            throw new IllegalStateException("test failed", getCause());
        }
    }

public abstract class TestRunnable implements Runnable {

    @Override
    public final void run() {
        try {
            execute();
        } catch (Exception e) {
            handleException(e);
        }
    }

    protected abstract void handleException(Throwable e);

    protected abstract void execute() throws Exception;

}

public abstract class AsyncThreadsTestRunnable extends TestRunnable {

    @Override
    protected final void handleException(Throwable e) {
        setCause(e);
    }

}

public class PlainSocketClient extends AsyncThreadsTestRunnable {

    @Override
    protected void execute() throws Exception {
        final long startMillis = System.currentTimeMillis();
        try (Socket sock = new Socket(SRV_HOST, SRV_PORT)) {
            sock.getOutputStream().write(MSG);
        }
        NettyServerBossTest.this.actual.incrementAndGet();
        System.out.println("Client done. Duration: " + (System.currentTimeMillis() - startMillis));
    }

}

public class DiscardServerHandler extends SimpleChannelHandler {

    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
        NettyServerBossTest.this.channels.add(e.getChannel());
    }

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
        // discard
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
        e.getCause().printStackTrace();

        Channel ch = e.getChannel();
        ch.close();
    }
}

}

Answer 1

Creo que el tiempo que registra no gasta todo en socket abierto, gasta por cambio de hilo porque cuando el hilo A abre un socket, entonces la CPU puede cambiar al hilo B, luego cuando se cierra el socket, la CPU puede no cambiar al hilo A en una vez pero después de ejecutar muchos otros hilos. He cambiado su PlainSocketClient para agregar un sincronizado para asegurarse de que haya menos efecto de cambio de hilo:

public class PlainSocketClient extends AsyncThreadsTestRunnable {
    private static final String LOCK = "LOCK";

    @Override
    protected void execute() throws Exception {
        synchronized (LOCK) {
            final long startMillis = System.currentTimeMillis();
            try (Socket sock = new Socket(SRV_HOST, SRV_PORT)) {
                sock.getOutputStream().write(MSG);
            }
            NettyServerBossTest.this.actual.incrementAndGet();
            System.out.println("Client done. Duration: " + (System.currentTimeMillis() - startMillis));
        }
    }
}

luego, casi solo emiten 0 o 1. Puede hacerse una prueba usted mismo. Simplemente prueba el tiempo que lleva al cambiar de hilo, no significa que deba agregar un sincronizado en su código.