openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/TcpHandler.java

   1 /*
   2  * Copyright (c) 2013 Pantheon Technologies s.r.o. and others. All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.openflowjava.protocol.impl.core;
  10
  11 import io.netty.bootstrap.ServerBootstrap;
  12 import io.netty.buffer.PooledByteBufAllocator;
  13 import io.netty.channel.ChannelFuture;
  14 import io.netty.channel.ChannelOption;
  15 import io.netty.channel.nio.NioEventLoopGroup;
  16 import io.netty.channel.socket.nio.NioServerSocketChannel;
  17 import io.netty.handler.logging.LogLevel;
  18 import io.netty.handler.logging.LoggingHandler;
  19 import io.netty.util.concurrent.GenericFutureListener;
  20
  21 import java.net.InetAddress;
  22 import java.net.InetSocketAddress;
  23
  24 import org.opendaylight.openflowjava.protocol.api.connection.ThreadConfiguration;
  25 import org.slf4j.Logger;
  26 import org.slf4j.LoggerFactory;
  27
  28 import com.google.common.util.concurrent.ListenableFuture;
  29 import com.google.common.util.concurrent.SettableFuture;
  30
  31 /**
  32  * Class implementing server over TCP / TLS for handling incoming connections.
  33  *
  34  * @author michal.polkorab
  35  */
  36 public class TcpHandler implements ServerFacade {
  37     /*
  38      * High/low write watermarks, in KiB.
  39      */
  40     private static final int DEFAULT_WRITE_HIGH_WATERMARK = 64;
  41     private static final int DEFAULT_WRITE_LOW_WATERMARK = 32;
  42     /*
  43      * Write spin count. This tells netty to immediately retry a non-blocking
  44      * write this many times before moving on to selecting.
  45      */
  46     private static final int DEFAULT_WRITE_SPIN_COUNT = 16;
  47
  48     private static final Logger LOGGER = LoggerFactory.getLogger(TcpHandler.class);
  49
  50     private int port;
  51     private String address;
  52     private final InetAddress startupAddress;
  53     private NioEventLoopGroup workerGroup;
  54     private NioEventLoopGroup bossGroup;
  55     private final SettableFuture<Boolean> isOnlineFuture;
  56     private ThreadConfiguration threadConfig;
  57
  58     private TcpChannelInitializer channelInitializer;
  59
  60     /**
  61      * Constructor of TCPHandler that listens on selected port.
  62      *
  63      * @param port listening port of TCPHandler server
  64      */
  65     public TcpHandler(final int port) {
  66         this(null, port);
  67     }
  68
  69     /**
  70      * Constructor of TCPHandler that listens on selected address and port.
  71      * @param address listening address of TCPHandler server
  72      * @param port listening port of TCPHandler server
  73      */
  74     public TcpHandler(final InetAddress address, final int port) {
  75         this.port = port;
  76         this.startupAddress = address;
  77         isOnlineFuture = SettableFuture.create();
  78     }
  79
  80     /**
  81      * Starts server on selected port.
  82      */
  83     @Override
  84     public void run() {
  85         /*
  86          * We generally do not perform IO-unrelated tasks, so we want to have
  87          * all outstanding tasks completed before the executing thread goes
  88          * back into select.
  89          *
  90          * Any other setting means netty will measure the time it spent selecting
  91          * and spend roughly proportional time executing tasks.
  92          */
  93         workerGroup.setIoRatio(100);
  94
  95         final ChannelFuture f;
  96         try {
  97             ServerBootstrap b = new ServerBootstrap();
  98             b.group(bossGroup, workerGroup)
  99                     .channel(NioServerSocketChannel.class)
 100                     .handler(new LoggingHandler(LogLevel.DEBUG))
 101                     .childHandler(channelInitializer)
 102                     .option(ChannelOption.SO_BACKLOG, 128)
 103                     .option(ChannelOption.SO_REUSEADDR, true)
 104                     .childOption(ChannelOption.SO_KEEPALIVE, true)
 105                     .childOption(ChannelOption.TCP_NODELAY , true)
 106                     .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
 107                     .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, DEFAULT_WRITE_HIGH_WATERMARK * 1024)
 108                     .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, DEFAULT_WRITE_LOW_WATERMARK * 1024)
 109                     .childOption(ChannelOption.WRITE_SPIN_COUNT, DEFAULT_WRITE_SPIN_COUNT);
 110
 111             if (startupAddress != null) {
 112                 f = b.bind(startupAddress.getHostAddress(), port).sync();
 113             } else {
 114                 f = b.bind(port).sync();
 115             }
 116         } catch (InterruptedException e) {
 117             LOGGER.error("Interrupted while binding port {}", port, e);
 118             return;
 119         }
 120
 121         try {
 122             InetSocketAddress isa = (InetSocketAddress) f.channel().localAddress();
 123             address = isa.getHostString();
 124
 125             // Update port, as it may have been specified as 0
 126             this.port = isa.getPort();
 127
 128             LOGGER.debug("address from tcphandler: {}", address);
 129             isOnlineFuture.set(true);
 130             LOGGER.info("Switch listener started and ready to accept incoming tcp/tls connections on port: {}", port);
 131             f.channel().closeFuture().sync();
 132         } catch (InterruptedException e) {
 133             LOGGER.error("Interrupted while waiting for port {} shutdown", port, e);
 134         } finally {
 135             shutdown();
 136         }
 137     }
 138
 139     /**
 140      * Shuts down {@link TcpHandler}}
 141      */
 142     @Override
 143     public ListenableFuture<Boolean> shutdown() {
 144         final SettableFuture<Boolean> result = SettableFuture.create();
 145         workerGroup.shutdownGracefully();
 146         // boss will shutdown as soon, as worker is down
 147         bossGroup.shutdownGracefully().addListener(new GenericFutureListener<io.netty.util.concurrent.Future<Object>>() {
 148
 149             @Override
 150             public void operationComplete(
 151                     final io.netty.util.concurrent.Future<Object> downResult) throws Exception {
 152                 result.set(downResult.isSuccess());
 153                 if (downResult.cause() != null) {
 154                     result.setException(downResult.cause());
 155                 }
 156             }
 157
 158         });
 159         return result;
 160     }
 161
 162     /**
 163      *
 164      * @return number of connected clients / channels
 165      */
 166     public int getNumberOfConnections() {
 167         return channelInitializer.size();
 168     }
 169
 170     @Override
 171     public ListenableFuture<Boolean> getIsOnlineFuture() {
 172         return isOnlineFuture;
 173     }
 174
 175     /**
 176      * @return the port
 177      */
 178     public int getPort() {
 179         return port;
 180     }
 181
 182     /**
 183      * @return the address
 184      */
 185     public String getAddress() {
 186         return address;
 187     }
 188
 189     /**
 190      * @param channelInitializer
 191      */
 192     public void setChannelInitializer(TcpChannelInitializer channelInitializer) {
 193         this.channelInitializer = channelInitializer;
 194     }
 195
 196     @Override
 197     public void setThreadConfig(ThreadConfiguration threadConfig) {
 198         this.threadConfig = threadConfig;
 199     }
 200
 201     /**
 202      * Initiate event loop groups
 203      * @param threadConfiguration number of threads to be created, if not specified in threadConfig
 204      */
 205     public void initiateEventLoopGroups(ThreadConfiguration threadConfiguration) {
 206         if (threadConfiguration != null) {
 207             bossGroup = new NioEventLoopGroup(threadConfiguration.getBossThreadCount());
 208             workerGroup = new NioEventLoopGroup(threadConfiguration.getWorkerThreadCount());
 209         } else {
 210             bossGroup = new NioEventLoopGroup();
 211             workerGroup = new NioEventLoopGroup();
 212         }
 213     }
 214
 215     /**
 216      * @return workerGroup
 217      */
 218     public NioEventLoopGroup getWorkerGroup() {
 219         return workerGroup;
 220     }
 221
 222 }