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