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.ServerSocketChannel;
  17 import io.netty.channel.socket.nio.NioServerSocketChannel;
  18 import io.netty.handler.logging.LogLevel;
  19 import io.netty.handler.logging.LoggingHandler;
  20 import io.netty.util.concurrent.GenericFutureListener;
  21
  22 import io.netty.channel.epoll.Epoll;
  23 import io.netty.channel.EventLoopGroup;
  24 import io.netty.channel.epoll.EpollEventLoopGroup;
  25 import io.netty.channel.epoll.EpollServerSocketChannel;
  26
  27 import java.net.InetAddress;
  28 import java.net.InetSocketAddress;
  29
  30 import org.opendaylight.openflowjava.protocol.api.connection.ThreadConfiguration;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 import com.google.common.util.concurrent.ListenableFuture;
  35 import com.google.common.util.concurrent.SettableFuture;
  36
  37 /**
  38  * Class implementing server over TCP / TLS for handling incoming connections.
  39  *
  40  * @author michal.polkorab
  41  */
  42 public class TcpHandler implements ServerFacade {
  43     /*
  44      * High/low write watermarks, in KiB.
  45      */
  46     private static final int DEFAULT_WRITE_HIGH_WATERMARK = 64;
  47     private static final int DEFAULT_WRITE_LOW_WATERMARK = 32;
  48     /*
  49      * Write spin count. This tells netty to immediately retry a non-blocking
  50      * write this many times before moving on to selecting.
  51      */
  52     private static final int DEFAULT_WRITE_SPIN_COUNT = 16;
  53
  54     private static final Logger LOG = LoggerFactory.getLogger(TcpHandler.class);
  55
  56     private int port;
  57     private String address;
  58     private final InetAddress startupAddress;
  59     private EventLoopGroup workerGroup;
  60     private EventLoopGroup bossGroup;
  61     private final SettableFuture<Boolean> isOnlineFuture;
  62     private ThreadConfiguration threadConfig;
  63
  64     private TcpChannelInitializer channelInitializer;
  65
  66     private Class<? extends ServerSocketChannel> socketChannelClass;
  67
  68     /**
  69      * Constructor of TCPHandler that listens on selected port.
  70      *
  71      * @param port listening port of TCPHandler server
  72      */
  73     public TcpHandler(final int port) {
  74         this(null, port);
  75     }
  76
  77     /**
  78      * Constructor of TCPHandler that listens on selected address and port.
  79      * @param address listening address of TCPHandler server
  80      * @param port listening port of TCPHandler server
  81      */
  82     public TcpHandler(final InetAddress address, final int port) {
  83         this.port = port;
  84         this.startupAddress = address;
  85         isOnlineFuture = SettableFuture.create();
  86     }
  87
  88     /**
  89      * Starts server on selected port.
  90      */
  91     @Override
  92     public void run() {
  93         /*
  94          * We generally do not perform IO-unrelated tasks, so we want to have
  95          * all outstanding tasks completed before the executing thread goes
  96          * back into select.
  97          *
  98          * Any other setting means netty will measure the time it spent selecting
  99          * and spend roughly proportional time executing tasks.
 100          */
 101         //workerGroup.setIoRatio(100);
 102
 103         final ChannelFuture f;
 104         try {
 105             ServerBootstrap b = new ServerBootstrap();
 106             b.group(bossGroup, workerGroup)
 107                     .channel(socketChannelClass)
 108                     .handler(new LoggingHandler(LogLevel.DEBUG))
 109                     .childHandler(channelInitializer)
 110                     .option(ChannelOption.SO_BACKLOG, 128)
 111                     .option(ChannelOption.SO_REUSEADDR, true)
 112                     .childOption(ChannelOption.SO_KEEPALIVE, true)
 113                     .childOption(ChannelOption.TCP_NODELAY , true)
 114                     .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
 115                     .childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, DEFAULT_WRITE_HIGH_WATERMARK * 1024)
 116                     .childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, DEFAULT_WRITE_LOW_WATERMARK * 1024)
 117                     .childOption(ChannelOption.WRITE_SPIN_COUNT, DEFAULT_WRITE_SPIN_COUNT);
 118
 119             if (startupAddress != null) {
 120                 f = b.bind(startupAddress.getHostAddress(), port).sync();
 121             } else {
 122                 f = b.bind(port).sync();
 123             }
 124         } catch (InterruptedException e) {
 125             LOG.error("Interrupted while binding port {}", port, e);
 126             return;
 127         }
 128
 129         try {
 130             InetSocketAddress isa = (InetSocketAddress) f.channel().localAddress();
 131             address = isa.getHostString();
 132
 133             // Update port, as it may have been specified as 0
 134             this.port = isa.getPort();
 135
 136             LOG.debug("address from tcphandler: {}", address);
 137             isOnlineFuture.set(true);
 138             LOG.info("Switch listener started and ready to accept incoming tcp/tls connections on port: {}", port);
 139             f.channel().closeFuture().sync();
 140         } catch (InterruptedException e) {
 141             LOG.error("Interrupted while waiting for port {} shutdown", port, e);
 142         } finally {
 143             shutdown();
 144         }
 145     }
 146
 147     /**
 148      * Shuts down {@link TcpHandler}}
 149      */
 150     @Override
 151     public ListenableFuture<Boolean> shutdown() {
 152         final SettableFuture<Boolean> result = SettableFuture.create();
 153         workerGroup.shutdownGracefully();
 154         // boss will shutdown as soon, as worker is down
 155         bossGroup.shutdownGracefully().addListener(new GenericFutureListener<io.netty.util.concurrent.Future<Object>>() {
 156
 157             @Override
 158             public void operationComplete(
 159                     final io.netty.util.concurrent.Future<Object> downResult) throws Exception {
 160                 result.set(downResult.isSuccess());
 161                 if (downResult.cause() != null) {
 162                     result.setException(downResult.cause());
 163                 }
 164             }
 165
 166         });
 167         return result;
 168     }
 169
 170     /**
 171      *
 172      * @return number of connected clients / channels
 173      */
 174     public int getNumberOfConnections() {
 175         return channelInitializer.size();
 176     }
 177
 178     @Override
 179     public ListenableFuture<Boolean> getIsOnlineFuture() {
 180         return isOnlineFuture;
 181     }
 182
 183     /**
 184      * @return the port
 185      */
 186     public int getPort() {
 187         return port;
 188     }
 189
 190     /**
 191      * @return the address
 192      */
 193     public String getAddress() {
 194         return address;
 195     }
 196
 197     /**
 198      * @param channelInitializer
 199      */
 200     public void setChannelInitializer(TcpChannelInitializer channelInitializer) {
 201         this.channelInitializer = channelInitializer;
 202     }
 203
 204     @Override
 205     public void setThreadConfig(ThreadConfiguration threadConfig) {
 206         this.threadConfig = threadConfig;
 207     }
 208
 209     /**
 210      * Initiate event loop groups
 211      * @param threadConfiguration number of threads to be created, if not specified in threadConfig
 212      */
 213     public void initiateEventLoopGroups(ThreadConfiguration threadConfiguration, boolean isEpollEnabled) {
 214
 215         if(isEpollEnabled) {
 216             initiateEpollEventLoopGroups(threadConfiguration);
 217         } else {
 218             initiateNioEventLoopGroups(threadConfiguration);
 219         }
 220     }
 221
 222     /**
 223      * Initiate Nio event loop groups
 224      * @param threadConfiguration number of threads to be created, if not specified in threadConfig
 225      */
 226     public void initiateNioEventLoopGroups(ThreadConfiguration threadConfiguration) {
 227         socketChannelClass = NioServerSocketChannel.class;
 228         if (threadConfiguration != null) {
 229             bossGroup = new NioEventLoopGroup(threadConfiguration.getBossThreadCount());
 230             workerGroup = new NioEventLoopGroup(threadConfiguration.getWorkerThreadCount());
 231         } else {
 232             bossGroup = new NioEventLoopGroup();
 233             workerGroup = new NioEventLoopGroup();
 234         }
 235         ((NioEventLoopGroup)workerGroup).setIoRatio(100);
 236     }
 237
 238     /**
 239      * Initiate Epoll event loop groups with Nio as fall back
 240      * @param threadConfiguration
 241      */
 242     protected void initiateEpollEventLoopGroups(ThreadConfiguration threadConfiguration) {
 243         try {
 244             socketChannelClass = EpollServerSocketChannel.class;
 245             if (threadConfiguration != null) {
 246                     bossGroup = new EpollEventLoopGroup(threadConfiguration.getBossThreadCount());
 247                 workerGroup = new EpollEventLoopGroup(threadConfiguration.getWorkerThreadCount());
 248             } else {
 249                 bossGroup = new EpollEventLoopGroup();
 250                 workerGroup = new EpollEventLoopGroup();
 251             }
 252             ((EpollEventLoopGroup)workerGroup).setIoRatio(100);
 253             return;
 254         } catch (Throwable ex) {
 255             LOG.debug("Epoll initiation failed");
 256         }
 257
 258         //Fallback mechanism
 259         initiateNioEventLoopGroups(threadConfiguration);
 260     }
 261
 262     /**
 263      * @return workerGroup
 264      */
 265     public EventLoopGroup getWorkerGroup() {
 266         return workerGroup;
 267     }
 268
 269 }