Showing posts with label socket. Show all posts
Showing posts with label socket. Show all posts

Tuesday, 11 October 2016

Performance comparison of multi-threaded client-server application between blocking and Non-blocking IO in java.

In this article I am going to talk about basic multi-threaded client-server application. We will create and observe behavior using blocking IO (OIO) and after that We will create similar application using NIO and observe the behavior and performance impacts.

1) Blocking IO multi-threaded client and server program

Server Code:
 import io.netty.util.concurrent.DefaultThreadFactory;   
  import java.io.BufferedReader;   
  import java.io.IOException;   
  import java.io.InputStreamReader;   
  import java.io.PrintWriter;   
  import java.net.ServerSocket;   
  import java.net.Socket;   
  import java.util.concurrent.ExecutorService;   
  import java.util.concurrent.Executors;   
     
  public class OioServer {   
   public static void main(String... args) throws IOException {   
    ExecutorService threadPool = Executors.newFixedThreadPool(5, new DefaultThreadFactory("serverpool"));   
    try (ServerSocket listener = new ServerSocket(6689)) {   
     while (true) {   
      // this statement remains blocked till the time any new connection request is received from client   
      Socket socket = listener.accept();   
      threadPool.execute(new MyServerThread(socket));   
     }   
    }   
   }   
   private static class MyServerThread implements Runnable {   
    private final Socket socket;   
     
    private MyServerThread(Socket socket) {   
     this.socket = socket;   
    }   
    public void run() {   
     try {   
      socket.setKeepAlive(true);   
      BufferedReader clientDataStream = new BufferedReader(new InputStreamReader(socket.getInputStream()));   
      PrintWriter out = new PrintWriter(socket.getOutputStream(), true);   
      // first message from client is client name   
      String clientName = clientDataStream.readLine();   
      //sending back acknowledgement to client   
      out.println(clientName + " connected");   
      System.out.println(clientName + " connected");   
      // second message from client is dummy data to process by server   
      System.out.println(clientDataStream.readLine() + " Processed by server");   
     } catch (IOException e) {   
      e.printStackTrace();   
     } finally {   
      try {   
       socket.close();   
      } catch (IOException e) {   
       e.printStackTrace();   
      }   
     }   
    }   
   }   
  }   




Let me explain what is happening in this program:

1) we are creating a thread pool of fixed size 5. As this is server you have to manage resources so you can not create infinite threads or cached thread pool which keeps on creating threads. Else the server will be out of processing and memory resources.For this example I have kept the count very low but in real life it could be much higher.

2) In the main thread we are running an infinite while loop which accepts a connection from client , once the client connection received it submits the handling of that connection to a thread .

3) As we have 5 threads only for processing actual requests , at a time only 5 client connections will be handled. If client request come and thread pool has no free thread than it queues up that thread. When ever any thread become free in thread pool the queued task will be assigned to that thread.

4) In the client handling thread , first we are expecting client name from client . Second we are sending acknowledgement back to the client. Third we are again expecting some data from client  to be processed by server. Than we just sysout that data in server.

5) clientDataStream.readLine() is a blocking operation and that method call remains blocked till the time some data is received from client. The processing thread remains occupied till the time that method call is not over.

Client Code:
 import io.netty.util.concurrent.DefaultThreadFactory;   
  import java.io.BufferedReader;   
  import java.io.IOException;   
  import java.io.InputStreamReader;   
  import java.io.PrintWriter;   
  import java.net.Socket;   
  import java.util.concurrent.ExecutorService;   
  import java.util.concurrent.Executors;   
     
  public class OioClient {   
   public static void main(String... args) throws IOException, InterruptedException {   
    int count = 1;   
    ExecutorService threadPool = Executors.newCachedThreadPool(new DefaultThreadFactory("clientpool"));   
    while (true) {   
     threadPool.execute(new MyClientThread(count));   
     if (++count > 10) break;   
     Thread.sleep(500); // just to make sure ordering  
   
    }   
   }   
     
   private static class MyClientThread implements Runnable {   
    private final int number;   
     
    private MyClientThread(int number) {   
     this.number = number;   
    }   
    public void run() {   
     Socket s = null;   
     try {   
      //trying to connect to server   
      s = new Socket("localhost", 6689);   
      s.setKeepAlive(true);   
      BufferedReader serverDataStream = new BufferedReader(new InputStreamReader(s.getInputStream()));   
      PrintWriter out = new PrintWriter(s.getOutputStream(), true);   
      //sending client name to server   
      out.println("client" + number);   
      //receive response from server   
      System.out.println(serverDataStream.readLine());   
      // sleep for 15 seconds   
      Thread.sleep(15000L);   
      //send some data to server for processing   
      out.println("client" + number + " Data");   
     } catch (Exception e) {   
      System.out.println("In the Exception for Client " + number);   
      e.printStackTrace();   
     } finally {   
      try {   
       if (s != null) {   
        s.close();   
       }   
      } catch (IOException e) {   
       e.printStackTrace();   
      }   
     }   
    }   
     
   }   
  }   
     

What we are doing here is:
1) In the main thread we are creating 10 client threads and submit them for processing.

2) Each client thread First connecting to server. Than sending its name to server. After that it expects response from server. Than it is sleeping for 15 seconds and after that sending some data to server for processing. Than the thread is over. It will close the socket and come out of execution.

3) As we have seen that ,At a time server can handle 5 threads only and here each client thread has 15 second sleep inside the client server communication. So each thread in the server code which handles client communication will take minimum 15 seconds or more to become free and available for processing some new client request.


Client outcome:

















As you can see after getting acknowledgement from first 5 threads it will take long pause because the remaining 5 threads submitted for execution but server has no threads available for processing that next 5 connection as it is busy handling the first 5 connection request from client.


Server outcome:






















As you can see when server receives first 5 connection from client it prints the client name and send acknowledgement to client and after that each server thread waits for client to send some data for processing , but client takes 15 seconds delay to send that data so all server threads remains in blocked state and no new client connection request will be accepted , all of the rest connection request threads will be queued up for processing. Once each processing threads receives response from client they are processing it and those threads will be available for queued connection request. Once all queued request filled up thread-pool again than all of them remains in blocked state till the time they receive processing data from client.


  // second message from client is dummy data to process by server   
     System.out.println(clientDataStream.readLine() + " Processed by server");   

Above lines from server code makes the threads block as client will send data after 15 seconds delay.

Now Lets see similar client server interaction with NIO using Netty framework.
I will keep all the parameters like thread pool size , sleep time exactly same as OIO program.


2) NIO multi-threaded client and server program using Netty

If you are new to Netty please read my previous article in which I gave basic introduction related to Netty.
http://techxperiment.blogspot.in/2016/09/demonstration-of-basic-difference.html

Server Code:
 import io.netty.bootstrap.ServerBootstrap;   
  import io.netty.channel.*;   
  import io.netty.channel.nio.NioEventLoopGroup;   
  import io.netty.channel.socket.SocketChannel;   
  import io.netty.channel.socket.nio.NioServerSocketChannel;   
  import io.netty.handler.codec.LineBasedFrameDecoder;   
  import io.netty.handler.codec.string.StringDecoder;   
  import io.netty.handler.codec.string.StringEncoder;   
  import io.netty.util.concurrent.DefaultThreadFactory;   
     
     
  public class NioServer {   
   public static void main(String... args) throws InterruptedException {   
    EventLoopGroup bossGroup = new NioEventLoopGroup(1); // 1 thread to accept connections   
    //5 threads to process the connections   
    EventLoopGroup workerGroup = new NioEventLoopGroup(5, new DefaultThreadFactory("serverpool"));   
    try {   
     ServerBootstrap b = new ServerBootstrap();   
     b.group(bossGroup, workerGroup)   
       .channel(NioServerSocketChannel.class)   
       .childHandler(new ChannelInitializer<SocketChannel>() {   
        @Override   
        public void initChannel(SocketChannel ch) throws Exception {   
         ch.pipeline().addLast(new LineBasedFrameDecoder(50));   
         ch.pipeline().addLast(new StringDecoder());   
         ch.pipeline().addLast(new StringEncoder());   
     
         ch.pipeline().addLast(new MyServerChannelHandler());   
        }   
       })   
       .childOption(ChannelOption.SO_KEEPALIVE, true);   
     // Bind and start to accept incoming connections on port 8881.   
     ChannelFuture f = b.bind(8881).sync();   
     f.addListener(channelFuture -> {   
      if (channelFuture.isSuccess()) {   
       System.out.println("Server started...");   
      } else {   
       System.out.println(channelFuture.cause());   
      }   
     });   
     f.channel().closeFuture().sync();   
    } finally {   
     workerGroup.shutdownGracefully();   
     bossGroup.shutdownGracefully();   
    }   
   }   
     
   private static class MyServerChannelHandler extends SimpleChannelInboundHandler<String> {   
    @Override   
    public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {   
     if (!msg.contains("Data")) { // if client sends its name   
      System.out.println(msg + " connected"); // print on console   
      ctx.writeAndFlush(msg + " connected" + "\n"); //sending back acknowledgement to client   
     } else { // if client sends some data for processing   
      System.out.println(msg + " Processed by server"); // print on console   
     }   
    }   
   }   
  }   


What we are doing here:

1) creating 1 thread which accepts the client connections and 5 threads which can actually process that request. Which is exactly same as OIO server code.

2)MyServerChannelHandler handles the client request exactly in similar way of OIO server program. I wrote comments to make it clear the response flow of server.


Client Code:
 import io.netty.bootstrap.Bootstrap;   
  import io.netty.channel.*;   
  import io.netty.channel.nio.NioEventLoopGroup;   
  import io.netty.channel.socket.SocketChannel;   
  import io.netty.channel.socket.nio.NioSocketChannel;   
  import io.netty.handler.codec.LineBasedFrameDecoder;   
  import io.netty.handler.codec.string.StringDecoder;   
  import io.netty.handler.codec.string.StringEncoder;   
  import io.netty.util.concurrent.DefaultThreadFactory;   
  import io.netty.util.concurrent.GenericFutureListener;   
  import java.util.concurrent.ExecutorService;   
  import java.util.concurrent.Executors;   
     
  public class NioClient {   
   public static void main(String... args) throws InterruptedException {   
    int count = 1;   
    ExecutorService threadPool = Executors.newCachedThreadPool(new DefaultThreadFactory("clientpool"));   
    while (true) {   
     threadPool.execute(new MyClientThread(count));   
     if (++count > 10) break;   
     Thread.sleep(500); // just to make sure ordering   
    }   
   }   
     
   private static class MyClientThread implements Runnable {   
    private final int number;   
     
    private MyClientThread(int number) {   
     this.number = number;   
    }   
    @Override   
    public void run() {   
     EventLoopGroup workerGroup = new NioEventLoopGroup(1);   
     try {   
      Bootstrap b = new Bootstrap();   
      b.group(workerGroup);   
      b.channel(NioSocketChannel.class);   
      b.option(ChannelOption.SO_KEEPALIVE, true);   
      b.handler(new ChannelInitializer<SocketChannel>() {   
       @Override   
       public void initChannel(SocketChannel ch) throws Exception {   
        ch.pipeline().addLast(new LineBasedFrameDecoder(50));   
        ch.pipeline().addLast(new StringDecoder());   
        ch.pipeline().addLast(new StringEncoder());   
        ch.pipeline().addLast(new MyClientChannelHandler());   
       }   
      });   
      ChannelFuture f = null;   
      try {   
       //Trying to connect to server   
       f = b.connect("localhost", 8881).sync();   
      } catch (InterruptedException e) {   
       e.printStackTrace();   
      }   
      f.addListener(new GenericFutureListener<ChannelFuture>() {   
       public void operationComplete(ChannelFuture channelFuture) throws Exception {   
        if (channelFuture.isSuccess()) {   
         // any logger statements when client connected to server   
        } else {   
         System.out.println(channelFuture.cause());   
        }   
       }   
      });   
      //sending client name to server   
      f.channel().writeAndFlush("client" + number + "\n");   
      // sleep for 15 seconds   
      Thread.sleep(15000);   
      //send some data to server for processing   
      f.channel().writeAndFlush("client" + number + " Data" + "\n");   
     
      // Wait until the connection is closed.   
      try {   
       f.channel().closeFuture().sync();   
      } catch (InterruptedException e) {   
       e.printStackTrace();   
      }   
     } catch (InterruptedException e) {   
      e.printStackTrace();   
     } finally {   
      workerGroup.shutdownGracefully();   
     }   
    }   
   }   
     
   private static class MyClientChannelHandler extends SimpleChannelInboundHandler<String> {   
    @Override   
    public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {   
     System.out.println(msg); // simply print on console whatever data received from client   
    }   
   }   
  }   

Here we are doing:
1) creating 10 threads which will try to connect to server and interact.

2) each thread once connected , first sends its name to client. Second sleep for 15 seconds , after that it sends data for processing to server.

3) MyClientChannelHandler will simply prints what ever response received from server.

Lets see the outcome of this program:

Client outcome:













Here we can see that all 10 clients get connected and received acknowledgement from server without any delay even though server has only 5 threads to process connections.
I will explain the reason very soon.

Server outcome:






















As you can see even though server has 5 threads only to handle client connection ,still it allows 10 client connection at a time. After that a long pause , as all client takes 15 seconds delay to send processing data . The key thing is server threads are not blocked and do not wait for any data from client. They will process it once it is available and mean while they can handle other client connections. This is the key aspect of NIO performance benefit over OIO.



 @Override   
    public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {   
     if (!msg.contains("Data")) { // if client sends its name   
      System.out.println(msg + " connected"); // print on console   
      ctx.writeAndFlush(msg + " connected" + "\n"); //sending back acknowledgement to client   
     } else { // if client sends some data for processing   
      System.out.println(msg + " Processed by server"); // print on console   
     }   
    }   


Above method does not block server threads and do not wait for any data from client. It will be asynchronously called when ever any data is available from client. Till that time it can utilize the same server threads for some completely different client connections.

In this article I have talked about NIO benefit on server side but similar thing you can think for client side also.

Now imagine a scenario of client server application where there are thousands of client concurrently accessing server and both of them have chain of operations like reading data from user, writing data in DB, calculate some business logic. Imagine if all the time server and client threads remains blocked than client can see significant delay in response from server and can not resume some other work till the time it receives outcome from response while on server side even though threads are idle and waiting for some data from client so they can not be utilized for other clients.
NIO is must and extremely powerful tool for high performance network applications.

I hope I am correctly able to make my point in this article.

Please post your comments and doubts!!! 

Friday, 30 September 2016

Demonstration of Basic difference between blocking and non-blocking IO in java using Netty framework.

In this article I am going to talk very briefly about the key difference between blocking and non-blocking IO operation.

First I am going to show you the simple socket programming using traditional blocking IO (OIO) and after that I will show you the non-blocking IO (NIO) example using Netty framework.


1) Blocking IO client and server program

Server code:
 public class OioServer {   
   public static void main(String... args) throws IOException {   
    ServerSocket listener = new ServerSocket(6689);   
    try {   
     while (true) {   
      //listening for client connection   
      Socket socket = listener.accept();   
      try {   
   
       //reading stream received from client   
       BufferedReader clientDataStream = new BufferedReader(new InputStreamReader(socket.getInputStream()));   
       String clientData = clientDataStream.readLine();   
       System.out.println("Client send :" + clientData);   
         
       //sending response to client        
       String response = clientData + " received by server";   
       PrintWriter out = new PrintWriter(socket.getOutputStream(), true);   
       out.println(response);   
      } finally {   
       socket.close();   
      }   
     }   
    } finally {   
     listener.close();   
    }   
   }   
  }   

Client code:
 public class OioClient {  
   public static void main(String... args) throws IOException {  
     while (true) {  
       BufferedReader br = new BufferedReader(new InputStreamReader(System.in));  
   
       //trying to connect to server   
       Socket s = new Socket("localhost", 6689);  
   
       //taking user input and sending it to server  
       PrintWriter out =  
           new PrintWriter(s.getOutputStream(), true);  
       System.out.print("Client input:");  
       out.println(br.readLine());  
   
       //reading back the server response   
       BufferedReader in = new BufferedReader(  
           new InputStreamReader(s.getInputStream()));  
       System.out.println(in.readLine());  
    }  
   
   }  
 }  


As you can see when I run this program it will run both client and server in a loop and keep taking input from client program and submitting it to the server while the server keep responding the client for each input received from client.Below is the output of server and client.

Client output:











Server output:







Lets make some changes to this code.
In the server code I am making below modifications.
I am adding 5 second sleep before server respond to client:
     //sending response to client    
     String response = clientData + " received by server";    
     PrintWriter out = new PrintWriter(socket.getOutputStream(), true);    
     try {    
     Thread.sleep(5000l);    
     } catch (InterruptedException e) {    
     e.printStackTrace();    
     }    
     out.println(response);  


In the client code I am making below changes:
Adding 2 sysout before and after fetching response from the server and calculating the delay in response.
     //reading back the server response   
     System.out.println("Waiting for server response...");   
     long startTime=System.nanoTime();   
     BufferedReader in = new BufferedReader(new InputStreamReader(s.getInputStream()));   
     System.out.println(in.readLine());   
     System.out.println("Server response received...Time taken:"+(System.nanoTime()-startTime)/1000000000.0+" seconds");   


Lets run the program and see the difference:

Client output:








As you can judge from output that the in.readLine() method will block the thread and return only when the server response received.
For that 5 second duration client thread remain in blocked state just waiting for that IO to complete.


2) Non-Blocking IO client and server program using Netty

For those who are not familiar with Netty let me give you a brief introduction.

From the Netty web site:
"Netty is a NIO client server framework which enables quick and easy development of network applications such as protocol servers and clients. It greatly simplifies and streamlines network programming such as TCP and UDP socket server."

It is a wrapper written over java NIO API for writing high performance client server application.
The next question comes into picture is what is java NIO ?? How it is different from the traditional IO(blocking IO). 

The answer of this question itself is a big discussion so I am giving you one of the best tutorial present on the web for NIO:

If you will go through this tutorial you can have basic idea about Channels, Buffers and Selectors.You will have understanding about how NIO is different from OIO by using Buffers to read from and write into rather than performing stream based OIO.
Netty is using java NIO API underneath and providing very powerful framework to write NIO based applications.
* You can also write OIO based application using Netty.


Server code:
 public class NioServer {   
     
   private static class MyServerChannelHandler extends SimpleChannelInboundHandler<String> {   
    @Override   
    public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {   
     System.out.println(msg + " received by server");   
     ctx.writeAndFlush("Hello from server\n");   
    }   
   }   
     
   public static void main(String... args) throws InterruptedException {   
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);   
    EventLoopGroup workerGroup = new NioEventLoopGroup(10);   
    try {   
     ServerBootstrap b = new ServerBootstrap();   
     b.group(bossGroup, workerGroup)   
       .channel(NioServerSocketChannel.class)   
       .childHandler(new ChannelInitializer<SocketChannel>() {   
        @Override   
        public void initChannel(SocketChannel ch) throws Exception {   
         ch.pipeline().addLast(new LineBasedFrameDecoder(100));   
         ch.pipeline().addLast(new StringDecoder());   
         ch.pipeline().addLast(new StringEncoder());   
     
         ch.pipeline().addLast(new MyServerChannelHandler());   
        }   
       })   
       .option(ChannelOption.SO_BACKLOG, 100)   
       .childOption(ChannelOption.SO_KEEPALIVE, true);   
     
     // Bind and start to accept incoming connections on port 8881.   
     ChannelFuture f = b.bind(8881).sync();   
     
     f.addListener(new GenericFutureListener<ChannelFuture>() {   
      public void operationComplete(ChannelFuture channelFuture) throws Exception {   
       if (channelFuture.isSuccess()) {   
        System.out.println("Server started...");   
       } else {   
        System.out.println(channelFuture.cause());   
       }   
      }   
     });   
     
     f.channel().closeFuture().sync();   
    } finally {   
     workerGroup.shutdownGracefully();   
     bossGroup.shutdownGracefully();   
    }   
     
   }   
  }    

Let me explain you what is happening in the above code by not going into much depth of syntactical part of Netty. We are exactly doing similar things which we did in OIO server program. We are trying to create a server socket on some port and listening on that for incoming connection of client and once client is connected we are responding it with "Hello from server\n". It is also sysout the message received from the client.

Client code:
 public class NioClient {  
   
   private static class MyClientChannelHandler extends SimpleChannelInboundHandler<String> {  
     @Override  
     public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {  
        System.out.println(msg);  
     }  
   }  
   
   public static void main(String... args) throws InterruptedException {  
     EventLoopGroup workerGroup = new NioEventLoopGroup();  
     try {  
       Bootstrap b = new Bootstrap();  
       b.group(workerGroup);  
       b.channel(NioSocketChannel.class);  
       b.option(ChannelOption.SO_KEEPALIVE, true);  
       b.handler(new ChannelInitializer<SocketChannel>() {  
         @Override  
         public void initChannel(SocketChannel ch) throws Exception {  
           ch.pipeline().addLast(new LineBasedFrameDecoder(100));  
           ch.pipeline().addLast(new StringDecoder());  
           ch.pipeline().addLast(new StringEncoder());  
           ch.pipeline().addLast(new MyClientChannelHandler());  
         }  
       });  
       // Start the client.  
       ChannelFuture f = b.connect("localhost", 8881).sync();  
   
       f.addListener(new GenericFutureListener<ChannelFuture>() {  
         public void operationComplete(ChannelFuture channelFuture) throws Exception {  
           if (channelFuture.isSuccess()) {  
             System.out.println("Client connected...");  
             channelFuture.channel().writeAndFlush("Hello World!!!\n");  
           } else {  
             System.out.println(channelFuture.cause());  
           }  
         }  
       });  
       // Wait until the connection is closed.  
       f.channel().closeFuture().sync();  
     } finally {  
       workerGroup.shutdownGracefully();  
     }  
   
   }  
 }  

The client code is trying to connect server on the specified port and once the connection is successful.it is sending "Hello World!!!\n" .  It is also sysout the response it receives from server.

Lets check out the outcome of the above code:

Client output:







Server output:






Lets make some changes to this code.
In the server code I am putting 5 second sleep before it respond to client.

 @Override  
     public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {  
       System.out.println(msg + " received by server");  
       Thread.sleep(5000l);  
       ctx.writeAndFlush("Hello from server\n");  
     }  


In the client code I am making below changes:
2 sysout before and after SENDING response to the server and calculate the delay in response.
Server respond back once it gets anything from client.
In Netty IO operations are asynchronous and we are handling server response in MyClientChannelHandler class. We print what ever is available from server when received.

  public void operationComplete(ChannelFuture channelFuture) throws Exception {  
           if (channelFuture.isSuccess()) {  
             System.out.println("Client connected...");  
             System.out.println("Waiting for server response...");  
             long startTime=System.nanoTime();  
             channelFuture.channel().writeAndFlush("Hello World!!!\n");  
             System.out.println("Server response received...Time taken:"+(System.nanoTime()-startTime)/1000000000.0+" seconds");  
           } else {  
             System.out.println(channelFuture.cause());  
           }  
         }  

Lets see the above change in action:
Client output:









You can see that channelFuture.channel().writeAndFlush("Hello World!!!\n"); writes to channel and returns immediately back.What ever returned by server will be handled in MyClientChannelHandler class when ever it will be available. The very important thing is the client thread will not be blocked and will not be waiting for server response in blocking manner. We are registering handlers with the channel. When ever any data available the client will handle it in non-blocking manner.

I will cover Netty programming in depth in future articles.

Please post your comments and doubts!!!