Netty ChunkWriteHandler is not working correctly - Not sending whole file only a few bytes of the file

[L_DisplayName]

Greetings All,

  I implementing Netty code to transfer a single 10MB binary file from one host (Node 0) to another host (Node 1) but only 8.5KB of the file gets transferred and I am having a hard time figuring out why.  I am using ChunkWriteHandler to send 1MB chunks of the file at a time via ChunkedNioFile (Please see code below).  In addition, I tried transferring files greater than 10MB, such as 100MB, 500MB and 1GB and only 8.5KB of the file is transferred.  If I reduce the chunk size specified in the ChunkedNioFile from 1MB to 512KB or lower, then 17 KB gets transferred which is double the size of previous file transfers. Also, I tried using just ChunkedFile but I received the same transfer results.  I can successfully transfer and receive the file headers: file name, file size (length) and the file offset (of where to start reading from or writting to), but only a few KB of the actual file.  Can someone tell me why ChunkWriteHandler will not send a complete file

Thank You,

Code Set up:

1. FileSenderInitializer.java
2. FileSenderHandler.java - 
3. FileSender.java     
4. FileReceiverInitializer.java
5. FileReceiverHandler.java 
6. FileReceiver.java

FileSenderInitializer.java - Initializes the channel pipeline with the channel handlers

public class FileSenderInitializer extends ChannelInitializer<SocketChannel> {

 @Override
 public void initChannel(SocketChannel ch) throws Exception {
 ch.pipeline().addLast(
  //new LengthFieldPrepender(8),
 new ChunkedWriteHandler(),
 new FileSenderHandler());
 }
}

FileSenderHandler.java - Sends the file header info - File Name, offset, length and then the Actual File

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
 try {
 String fileRequest = "ftp Node0/root/10MB_File.dat Node1/tmp/10MB_File_Copy.dat";

 //Source File to send / transfer to the Destination Node

 String theSrcFilePath =  "/root/10MB_File.dat";

 //File Name to write on the destination node, once the file is received  
 String theDestFilePath = "/tmp/10MB_File_Copy.dat";

//Get the source file to send
 File theFile = new File(theSrcFilePath);
 FileChannel theFileChannel = new RandomAccessFile(theFile, "r").getChannel();

//Get the length of the file
 long fileLength = theFileChannel.size();

 //Get the offset

 long offSet = 0;
 
 //Copy the offset to the ByteBuf
 ByteBuf offSetBuf = Unpooled.copyLong(offSet);
 //Copy the file length to the ByteBuf
 ByteBuf fileLengthBuf = Unpooled.copyLong(fileLength);

 //Get the Destination Filename (including the file path) in Bytes
 byte[] theDestFilePathInBytes = theDestFilePath.getBytes();
 //Get the length of theFilePath
 int theDestSize = theDestFilePathInBytes.length;
 //Copy the Dest File Path length to the ByteBuf
 ByteBuf theDestSizeBuf = Unpooled.copyInt(theDestSize);
 //Copy the theDestFilePathInBytes to the Byte Buf
 ByteBuf theDestFileBuf = Unpooled.copiedBuffer(theDestFilePathInBytes);
 

 //Send the file Headers: FileName Length, the FileName, the Offset and the file length
 ctx.write(theDestSizeBuf);
 ctx.write(theDestFileBuf);
 ctx.write(offSetBuf);
 ctx.write(fileLengthBuf);
 ctx.flush();

 //Send the 10MB File in 1MB chunks as specified by the following chunk size (1024*1024*1)
 ctx.write(new ChunkedNioFile(theFileChannel, offSet, fileLength, 1024 * 1024 * 1));
 ctx.flush();
 
 }catch(Exception e){
 System.err.printf("FileSenderHandler: Channel Active: Error: "+e.getMessage());
 e.printStackTrace();
 }
} //End channelActive

FileSender.java - Bootstraps the channel and connects this client/host to another host

public static void main(String[] args) throws Exception {
 // Configure the client/ File Sender
 EventLoopGroup group = new NioEventLoopGroup();
 try {
 Bootstrap b = new Bootstrap();
 b.group(group)
 .channel(NioSocketChannel.class)
 .option(ChannelOption.TCP_NODELAY, true)
 .handler(new FileSenderInitializer());

 // Start the client.
 ChannelFuture f = b.connect(HOST, PORT).sync();

 // Wait until the connection is closed.
 //f.channel().closeFuture().sync();
 } finally {
 // Shut down the event loop to terminate all threads.
 group.shutdownGracefully();
 }
 }
}

FileReceiverInitializer.java - Initializes the channel pipeline with the channel handlers

public class FileReceiverInitializer extends ChannelInitializer<SocketChannel> {

 public FileReceiverInitializer(){
 
 }

 @Override
 public void initChannel(SocketChannel ch) throws Exception {
 ch.pipeline().addLast( 
  //Read in 1MB data at a time (which is the max frame length), length field offset starts at 0, length of the length field is 8 bits, length adjustment is 0, strip the 8 bits representing the length field from the frame
 //new LengthFieldBasedFrameDecoder(1024*1024*1, 0, 8, 0, 8),
 new FileReceiverHandler());
 }
}

FileReceiverHandler.java - Receives the file header info - File Name, offset, length and then the actual file
public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
     while (msg.readableBytes() >= 1){
     //Read in the size of the File Name and it's directory path
     if (!fileNameStringSizeSet) {
         fileNameStringSizeBuf.writeBytes(msg, ((fileNameStringSizeBuf.writableBytes() >= msg.readableBytes()) ? msg.readableBytes() :   fileNameStringSizeBuf.writableBytes())); //INT_SIZE = 4 & LONG_SIZE = 8 (the byte size of an int and long)
         if (fileNameStringSizeBuf.readableBytes() >= INT_SIZE) {
             fileNameStringSize = fileNameStringSizeBuf.getInt(fileNameStringSizeBuf.readerIndex());//Get Size at index = 0;
             fileNameStringSizeSet = true;
             //Allocate a byteBuf to read in the actual file name and it's directory path
             fileNameStringBuf = ctx.alloc().buffer(fileNameStringSize);
         }
     } else if (!readInFileNameString) {
         //Read in the actual file name and it's corresponding directory path
         fileNameStringBuf.writeBytes(msg, ((fileNameStringBuf.writableBytes() >= msg.readableBytes()) ? msg.readableBytes() : fileNameStringBuf.writableBytes()));
         if (fileNameStringBuf.readableBytes() >= fileNameStringSize) {
             readInFileNameString = true;
             //convert the data in the fileNameStringBuf to an ascii string
             thefileName = fileNameStringBuf.toString(Charset.forName("US-ASCII"));
 
             //Create file
             emptyFile = new File(thefileName); //file Name includes the directory path
             f = new RandomAccessFile(emptyFile, "rw");
             fc = f.getChannel();
         }
     }else if (!readInOffset) {
         offSetBuf.writeBytes(msg, ((offSetBuf.writableBytes() >= msg.readableBytes()) ? msg.readableBytes() : offSetBuf.writableBytes()));
         if (offSetBuf.readableBytes() >= LONG_SIZE) {
             currentOffset = offSetBuf.getLong(offSetBuf.readerIndex());//Get Size at index = 0;
             readInOffset = true;
         }
     } else if (!readInFileLength) {
         fileLengthBuf.writeBytes(msg, ((fileLengthBuf.writableBytes() >= msg.readableBytes()) ? msg.readableBytes() : fileLengthBuf.writableBytes()));
         //LONG_SIZE = 8
         if (fileLengthBuf.readableBytes() >= LONG_SIZE) {
             fileLength = fileLengthBuf.getLong(fileLengthBuf.readerIndex());//Get Size at index = 0;
             remainingFileLength = fileLength;
             readInFragmentLength = true;
         }
     } else {
         if (!readInCompleteFile) {
             if (msg.readableBytes() < remainingFileLength) {
                 if (msg.readableBytes() > 0) {
                     currentFileBytesWrote = 0
                     while ( msg.readableBytes >= 1 ){
                         int fileBytesWrote = fc.write(msg.nioBuffer(msg.readerIndex(), msg.readableBytes()), currentOffset);
                         currentOffset += fileBytesWrote;
                         remainingFileLength -= fileBytesWrote;
                         msg.readerIndex(msg.readerIndex + fileBytesWrote);
                     }
                 }
             } else {
                 int remainingFileLengthInt = (int) remainingFileLength;
                 while (remainingFileLength >= 1){
                     int fileBytesWrote = fc.write(msg.nioBuffer(msg.readerIndex(), remainingFileLengthInt), currentOffset);

                     currentOffset += fileBytesWrote ;
                     remainingFileLength -= fileBytesWrote;
                     remainingFileLengthInt-= fileBytesWrote;
                     msg.readerIndex(msg.readerIndex + fileBytesWrote );
                }
 
               //Set readInCompleteFile to true
              readInCompleteFile = true;
 
             }
     }//End else if file chunk
     }//End Else
 }//End While
}//End Read Method


FileReceiver.java - Bootstraps the Server and accepts connections
public static void main(String[] args) throws Exception {
 // Configure the server
 EventLoopGroup bossGroup = new NioEventLoopGroup(1);
 EventLoopGroup workerGroup = new NioEventLoopGroup();
 try {
 ServerBootstrap b = new ServerBootstrap();
 b.group(bossGroup, workerGroup)
 .channel(NioServerSocketChannel.class)
 .handler(new LoggingHandler(LogLevel.INFO))
 .childHandler(new FileReceiverInitializer())
 .childOption(ChannelOption.AUTO_READ, true) 
 .bind(LOCAL_PORT).sync().channel().closeFuture().sync();
 } finally {
 bossGroup.shutdownGracefully();
 workerGroup.shutdownGracefully();
 }
}

[Tim Boudreau]

Because you commented out the line that keeps your program from exiting while it's still transferring data:

 // Wait until the connection is closed.
 //f.channel().closeFuture().sync();

Once you uncomment it, of course the program won't exit. See my answer to another recent thread in the last hour on how to solve that.

BTW, this line suggests you're doing development and running code you're developing as root??? Don't do that. Ever. Seriously. One accidental write to some part of your OS and you'll be reinstalling.

 String theSrcFilePath =  "/root/10MB_File.dat";

-Tim