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.opendaylight.openflowjava.protocol.impl.connection.ServerFacade;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29 import com.google.common.util.concurrent.ListenableFuture;
  30 import com.google.common.util.concurrent.SettableFuture;
  31
  32 /**
  33  * Class implementing server over TCP for handling incoming connections.
  34  *
  35  * @author michal.polkorab
  36  */
  37 public class TcpHandler implements ServerFacade {
  38     /*
  39      * High/low write watermarks, in KiB.
  40      */
  41     private static final int DEFAULT_WRITE_HIGH_WATERMARK = 64;
  42     private static final int DEFAULT_WRITE_LOW_WATERMARK = 32;
  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 LOGGER = LoggerFactory.getLogger(TcpHandler.class);
  50
  51     private int port;
  52     private String address;
  53     private final InetAddress startupAddress;
  54     private NioEventLoopGroup workerGroup;
  55     private NioEventLoopGroup bossGroup;
  56     private final SettableFuture<Boolean> isOnlineFuture;
  57     private ThreadConfiguration threadConfig;
  58
  59     private PublishingChannelInitializer channelInitializer;
  60
  61     /**
  62      * Constructor of TCPHandler that listens on selected port.
  63      *
  64      * @param port listening port of TCPHandler server
  65      */
  66     public TcpHandler(final int port) {
  67         this(null, port);
  68     }
  69
  70     /**
  71      * Constructor of TCPHandler that listens on selected address and port.
  72      * @param address listening address of TCPHandler server
  73      * @param port listening port of TCPHandler server
  74      */
  75     public TcpHandler(final InetAddress address, final int port) {
  76         this.port = port;
  77         this.startupAddress = address;
  78         isOnlineFuture = SettableFuture.create();
  79     }
  80
  81     /**
  82      * Starts server on selected port.
  83      */
  84     @Override
  85     public void run() {
  86         if (threadConfig != null) {
  87             bossGroup = new NioEventLoopGroup(threadConfig.getBossThreadCount());
  88             workerGroup = new NioEventLoopGroup(threadConfig.getWorkerThreadCount());
  89         } else {
  90             bossGroup = new NioEventLoopGroup();
  91             workerGroup = new NioEventLoopGroup();
  92         }
  93
  94         /*
  95          * We generally do not perform IO-unrelated tasks, so we want to have
  96          * all outstanding tasks completed before the executing thread goes
  97          * back into select.
  98          *
  99          * Any other setting means netty will measure the time it spent selecting
 100          * and spend roughly proportional time executing tasks.
 101          */
 102         workerGroup.setIoRatio(100);
 103
 104         final ChannelFuture f;
 105         try {
 106             ServerBootstrap b = new ServerBootstrap();
 107             b.group(bossGroup, workerGroup)
 108                     .channel(NioServerSocketChannel.class)
 109                     .handler(new LoggingHandler(LogLevel.DEBUG))
 110                     .childHandler(channelInitializer)
 111                     .option(ChannelOption.SO_BACKLOG, 128)
 112                     .option(ChannelOption.SO_REUSEADDR, true)
 113                     .childOption(ChannelOption.SO_KEEPALIVE, 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             LOGGER.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             LOGGER.debug("address from tcphandler: {}", address);
 137             isOnlineFuture.set(true);
 138             LOGGER.info("Switch listener started and ready to accept incoming connections on port: {}", port);
 139             f.channel().closeFuture().sync();
 140         } catch (InterruptedException e) {
 141             LOGGER.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(PublishingChannelInitializer channelInitializer) {
 201         this.channelInitializer = channelInitializer;
 202     }
 203
 204     /**
 205      * @param threadConfig EventLoopGroup configuration
 206      */
 207     public void setThreadConfig(ThreadConfiguration threadConfig) {
 208         this.threadConfig = threadConfig;
 209     }
 210 }