Store connection on channel to avoid hash lookup every time

src/main/java/io/vertx/core/datagram/impl/DatagramServerHandler.java

@@ -17,31 +17,40 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.socket.DatagramPacket;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.impl.ContextImpl;
 import io.vertx.core.net.impl.VertxHandler;
 
-import java.util.HashMap;
-
 /**
  * @author <a href="mailto:nmaurer@redhat.com">Norman Maurer</a>
  */
 final class DatagramServerHandler extends VertxHandler<DatagramSocketImpl> {
-  private final DatagramSocketImpl server;
 
-  DatagramServerHandler(DatagramSocketImpl server) {
-        super(new HashMap<>());
-    this.server = server;
+  private final DatagramSocketImpl socket;
+
+  DatagramServerHandler(DatagramSocketImpl socket) {
+    this.socket = socket;
   }
 
   @Override
   public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
     super.handlerAdded(ctx);
-    connectionMap.put(ctx.channel(), server);
   }
 
+  @Override
+  protected DatagramSocketImpl getConnection(Channel channel) {
+    return socket;
+  }
+
+  @Override
+  protected DatagramSocketImpl removeConnection(Channel channel) {
+    return socket;
+  }
+
+
   @SuppressWarnings("unchecked")
   @Override
   protected void channelRead(final DatagramSocketImpl server, final ContextImpl context, ChannelHandlerContext chctx, final Object msg) throws Exception {

src/main/java/io/vertx/core/http/impl/HttpClientImpl.java

@@ -17,47 +17,29 @@
 package io.vertx.core.http.impl;
 
 import io.netty.bootstrap.Bootstrap;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelFutureListener;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.ChannelPipeline;
-import io.netty.channel.FixedRecvByteBufAllocator;
-import io.netty.channel.socket.nio.NioSocketChannel;
-import io.netty.handler.codec.http.HttpClientCodec;
-import io.netty.handler.codec.http.HttpContent;
-import io.netty.handler.codec.http.HttpContentDecompressor;
-import io.netty.handler.codec.http.HttpResponse;
-import io.netty.handler.codec.http.LastHttpContent;
+import io.netty.channel.*;
+import io.netty.handler.codec.http.*;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.VertxException;
 import io.vertx.core.buffer.Buffer;
-import io.vertx.core.http.HttpClient;
-import io.vertx.core.http.HttpClientOptions;
-import io.vertx.core.http.HttpClientRequest;
-import io.vertx.core.http.HttpClientResponse;
+import io.vertx.core.http.*;
 import io.vertx.core.http.HttpMethod;
-import io.vertx.core.http.WebSocket;
-import io.vertx.core.http.WebSocketStream;
-import io.vertx.core.http.WebsocketVersion;
 import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
 import io.vertx.core.http.impl.ws.WebSocketFrameInternal;
 import io.vertx.core.impl.Closeable;
 import io.vertx.core.impl.ContextImpl;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
-import io.vertx.core.spi.metrics.Metrics;
-import io.vertx.core.spi.metrics.HttpClientMetrics;
 import io.vertx.core.net.impl.KeyStoreHelper;
 import io.vertx.core.net.impl.PartialPooledByteBufAllocator;
 import io.vertx.core.net.impl.SSLHelper;
+import io.vertx.core.spi.metrics.HttpClientMetrics;
+import io.vertx.core.spi.metrics.Metrics;
 import io.vertx.core.spi.metrics.MetricsProvider;
 
 import javax.net.ssl.SSLHandshakeException;
@@ -647,7 +629,7 @@ private void internalConnect(ContextImpl context, int port, String host, Handler
                                Handler<Throwable> connectErrorHandler, ConnectionLifeCycleListener listener) {
     Bootstrap bootstrap = new Bootstrap();
     bootstrap.group(context.eventLoop());
-    bootstrap.channel(NioSocketChannel.class);
+    bootstrap.channelFactory(new VertxNioSocketChannelFactory());
     sslHelper.validate(vertx);
     bootstrap.handler(new ChannelInitializer<Channel>() {
       @Override

src/main/java/io/vertx/core/http/impl/HttpServerImpl.java

@@ -21,7 +21,6 @@
 import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.group.ChannelGroupFuture;
 import io.netty.channel.group.DefaultChannelGroup;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.codec.http.*;
 import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
@@ -184,7 +183,7 @@ public synchronized HttpServer listen(int port, String host, Handler<AsyncResult
         serverChannelGroup = new DefaultChannelGroup("vertx-acceptor-channels", GlobalEventExecutor.INSTANCE);
         ServerBootstrap bootstrap = new ServerBootstrap();
         bootstrap.group(availableWorkers);
-        bootstrap.channel(NioServerSocketChannel.class);
+        bootstrap.channelFactory(new VertxNioServerChannelFactory());
         applyConnectionOptions(bootstrap);
         sslHelper.validate(vertx);
         bootstrap.childHandler(new ChannelInitializer<Channel>() {

src/main/java/io/vertx/core/http/impl/VertxHttpHandler.java

@@ -39,12 +39,39 @@
  */
 public abstract class VertxHttpHandler<C extends ConnectionBase> extends VertxHandler<C> {
 
+  private static ByteBuf safeBuffer(ByteBufHolder holder, ByteBufAllocator allocator) {
+    return safeBuffer(holder.content(), allocator);
+  }
+
+  protected Map<Channel, C> connectionMap;
+
   protected VertxHttpHandler(Map<Channel, C> connectionMap) {
-    super(connectionMap);
+    this.connectionMap = connectionMap;
   }
 
-  private static ByteBuf safeBuffer(ByteBufHolder holder, ByteBufAllocator allocator) {
-    return safeBuffer(holder.content(), allocator);
+  @Override
+  protected C getConnection(Channel channel) {
+    @SuppressWarnings("unchecked")
+    VertxNioSocketChannel<C> vch = (VertxNioSocketChannel<C>)channel;
+    // As an optimisation we store the connection on the channel - this prevents a lookup every time
+    // an event from Netty arrives
+    if (vch.conn != null) {
+      return vch.conn;
+    } else {
+      C conn = connectionMap.get(channel);
+      if (conn != null) {
+        vch.conn = conn;
+      }
+      return conn;
+    }
+  }
+
+  @Override
+  protected C removeConnection(Channel channel) {
+    @SuppressWarnings("unchecked")
+    VertxNioSocketChannel<C> vch = (VertxNioSocketChannel<C>)channel;
+    vch.conn = null;
+    return connectionMap.remove(channel);
   }
 
   @Override

src/main/java/io/vertx/core/http/impl/VertxNioServerChannelFactory.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2014 Red Hat, Inc.
+ *
+ *  All rights reserved. This program and the accompanying materials
+ *  are made available under the terms of the Eclipse Public License v1.0
+ *  and Apache License v2.0 which accompanies this distribution.
+ *
+ *  The Eclipse Public License is available at
+ *  http://www.eclipse.org/legal/epl-v10.html
+ *
+ *  The Apache License v2.0 is available at
+ *  http://www.opensource.org/licenses/apache2.0.php
+ *
+ *  You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.vertx.core.http.impl;
+
+import io.netty.bootstrap.ChannelFactory;
+
+/**
+ * @author <a href="http://tfox.org">Tim Fox</a>
+ */
+public class VertxNioServerChannelFactory implements ChannelFactory<VertxNioServerSocketChannel> {
+
+  @Override
+  public VertxNioServerSocketChannel newChannel() {
+    return new VertxNioServerSocketChannel();
+  }
+}

src/main/java/io/vertx/core/http/impl/VertxNioServerSocketChannel.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2014 Red Hat, Inc.
+ *
+ *  All rights reserved. This program and the accompanying materials
+ *  are made available under the terms of the Eclipse Public License v1.0
+ *  and Apache License v2.0 which accompanies this distribution.
+ *
+ *  The Eclipse Public License is available at
+ *  http://www.eclipse.org/legal/epl-v10.html
+ *
+ *  The Apache License v2.0 is available at
+ *  http://www.opensource.org/licenses/apache2.0.php
+ *
+ *  You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.vertx.core.http.impl;
+
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.vertx.core.logging.Logger;
+import io.vertx.core.logging.LoggerFactory;
+
+import java.nio.channels.SocketChannel;
+import java.util.List;
+
+/**
+ * @author <a href="http://tfox.org">Tim Fox</a>
+ */
+public class VertxNioServerSocketChannel extends NioServerSocketChannel {
+
+  private static final Logger log = LoggerFactory.getLogger(VertxNioServerSocketChannel.class);
+
+  @Override
+  protected int doReadMessages(List<Object> buf) throws Exception {
+    SocketChannel ch = javaChannel().accept();
+
+    try {
+      if (ch != null) {
+        buf.add(new VertxNioSocketChannel(this, ch));
+        return 1;
+      }
+    } catch (Throwable t) {
+      log.warn("Failed to create a new channel from an accepted socket.", t);
+
+      try {
+        ch.close();
+      } catch (Throwable t2) {
+        log.warn("Failed to close a socket.", t2);
+      }
+    }
+
+    return 0;
+  }
+}

src/main/java/io/vertx/core/http/impl/VertxNioSocketChannel.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2014 Red Hat, Inc.
+ *
+ *  All rights reserved. This program and the accompanying materials
+ *  are made available under the terms of the Eclipse Public License v1.0
+ *  and Apache License v2.0 which accompanies this distribution.
+ *
+ *  The Eclipse Public License is available at
+ *  http://www.eclipse.org/legal/epl-v10.html
+ *
+ *  The Apache License v2.0 is available at
+ *  http://www.opensource.org/licenses/apache2.0.php
+ *
+ *  You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.vertx.core.http.impl;
+
+import io.netty.channel.Channel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+
+import java.nio.channels.SocketChannel;
+
+/**
+ * We provide this class so we can store a reference to the connection on it.
+ *
+ * This means we don't have to do a lookup in the connectionMap for every message received which improves
+ * performance.
+ *
+ * @author <a href="http://tfox.org">Tim Fox</a>
+ */
+public class VertxNioSocketChannel<C> extends NioSocketChannel {
+
+  public C conn;
+
+  public VertxNioSocketChannel(Channel parent, SocketChannel socket) {
+    super(parent, socket);
+  }
+
+  public VertxNioSocketChannel() {
+  }
+}

src/main/java/io/vertx/core/http/impl/VertxNioSocketChannelFactory.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2014 Red Hat, Inc.
+ *
+ *  All rights reserved. This program and the accompanying materials
+ *  are made available under the terms of the Eclipse Public License v1.0
+ *  and Apache License v2.0 which accompanies this distribution.
+ *
+ *  The Eclipse Public License is available at
+ *  http://www.eclipse.org/legal/epl-v10.html
+ *
+ *  The Apache License v2.0 is available at
+ *  http://www.opensource.org/licenses/apache2.0.php
+ *
+ *  You may elect to redistribute this code under either of these licenses.
+ */
+
+package io.vertx.core.http.impl;
+
+import io.netty.bootstrap.ChannelFactory;
+
+/**
+ * @author <a href="http://tfox.org">Tim Fox</a>
+ */
+public class VertxNioSocketChannelFactory implements ChannelFactory<VertxNioSocketChannel> {
+
+  @Override
+  public VertxNioSocketChannel newChannel() {
+    return new VertxNioSocketChannel<>();
+  }
+}

src/main/java/io/vertx/core/net/impl/VertxHandler.java

@@ -26,18 +26,14 @@
 import io.netty.handler.timeout.IdleStateEvent;
 import io.vertx.core.impl.ContextImpl;
 
-import java.util.Map;
-
 /**
  * @author <a href="mailto:nmaurer@redhat.com">Norman Maurer</a>
  */
 public abstract class VertxHandler<C extends ConnectionBase> extends ChannelDuplexHandler {
 
-  protected final Map<Channel, C> connectionMap;
+  protected abstract C getConnection(Channel ch);
 
-  protected VertxHandler(Map<Channel, C> connectionMap) {
-    this.connectionMap = connectionMap;
-  }
+  protected abstract C removeConnection(Channel ch);
 
   protected ContextImpl getContext(C connection) {
     return connection.getContext();
@@ -66,7 +62,7 @@ protected static ByteBuf safeBuffer(ByteBuf buf, ByteBufAllocator allocator) {
   @Override
   public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
     Channel ch = ctx.channel();
-    C conn = connectionMap.get(ch);
+    C conn = getConnection(ch);
     if (conn != null) {
       ContextImpl context = getContext(conn);
       context.executeFromIO(conn::handleInterestedOpsChanged);
@@ -77,7 +73,7 @@ public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exceptio
   public void exceptionCaught(ChannelHandlerContext chctx, final Throwable t) throws Exception {
     Channel ch = chctx.channel();
     // Don't remove the connection at this point, or the handleClosed won't be called when channelInactive is called!
-    C connection = connectionMap.get(ch);
+    C connection = getConnection(ch);
     if (connection != null) {
       ContextImpl context = getContext(connection);
       context.executeFromIO(() -> {
@@ -97,7 +93,7 @@ public void exceptionCaught(ChannelHandlerContext chctx, final Throwable t) thro
   @Override
   public void channelInactive(ChannelHandlerContext chctx) throws Exception {
     Channel ch = chctx.channel();
-    C connection = connectionMap.remove(ch);
+    C connection = removeConnection(ch);
     if (connection != null) {
       ContextImpl context = getContext(connection);
       context.executeFromIO(connection::handleClosed);
@@ -106,7 +102,7 @@ public void channelInactive(ChannelHandlerContext chctx) throws Exception {
 
   @Override
   public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
-    C conn = connectionMap.get(ctx.channel());
+    C conn = getConnection(ctx.channel());
     if (conn != null) {
       ContextImpl context = getContext(conn);
       context.executeFromIO(conn::endReadAndFlush);
@@ -116,7 +112,7 @@ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
   @Override
   public void channelRead(ChannelHandlerContext chctx, Object msg) throws Exception {
     Object message = safeObject(msg, chctx.alloc());
-    C connection = connectionMap.get(chctx.channel());
+    C connection = getConnection(chctx.channel());
 
     ContextImpl context;
     if (connection != null) {