Merge pull request #9 from sakagg/multithread

Multithreading all the operations

Author: chandan5

src/Client/Client.java

@@ -10,15 +10,13 @@
 import NameNode.NameNode;
 import Proto.Hdfs;
 import Proto.ProtoMessage;
-import com.google.protobuf.ByteString;
 import com.google.protobuf.InvalidProtocolBufferException;
 import java.io.BufferedReader;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.FileReader;
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.MalformedURLException;
 import java.rmi.Naming;
@@ -36,67 +34,25 @@
  *
  * @author saksham
  */
-public class Client {
-    private static final String NN_NAME = "NameNode";
-    private static final String DN_PREFIX = "DataNode";
-    private static final Integer CHUNK_SIZE = 128*1024*1024;
-    private static final Properties props = new Properties();
+
+class ClientOperation implements Runnable {
 
-    private INameNode nn = null;
+    public enum OperationType { PUT, GET, LIST };
 
-    public static void log(String s) {
-        String op = String.valueOf(System.currentTimeMillis()) + " ";
-        op += "[Client] ";
-        op += ": ";
-        System.out.println(op + s);
-    }
+    private OperationType op;
+    private INameNode nn;
+    private String inputFilename;
+    private String outputFilename;
 
-    public static void main(String args[]) {
-        try {
-            props.load(new BufferedReader(new FileReader("config.properties")));
-        } catch (IOException ex) {
-            Logger.getLogger(NameNode.class.getName()).log(Level.SEVERE, null, ex);
-        }
-        
-        Client client = new Client();
-        client.findnn();
-        client.mainloop();
+    public ClientOperation(OperationType op, INameNode nn) {
+        this.op = op;
+        this.nn = nn;
     }
 
-    public void findnn() {
-        while(nn == null)
-        {
-            try {
-                nn = (INameNode) Naming.lookup("rmi://" + props.getProperty("rmi.namenode.ip")
-                        + ":" + props.getProperty("rmi.namenode.port") + "/" + NN_NAME);
-                log("Found Name Node");
-            } catch (NotBoundException | MalformedURLException | RemoteException ex) {
-                Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
-            }
-            if (nn == null)
-                try {
-                    Thread.sleep(1000);
-                } catch (Exception e) {}
-        }
-    }
-    
-    public void listFiles() {
-        byte[] inp = "*".getBytes();
-        try {
-            byte[] res = nn.list(inp);
-            Hdfs.ListFilesResponse listFilesResponse = Hdfs.ListFilesResponse.parseFrom(res);
-            if(listFilesResponse.getStatus() == 1) {
-                listFilesResponse.getFileNamesList().stream().forEach((fileName) -> {
-                    System.out.println(fileName);
-                });
-                System.out.println("");
-            } else {
-                System.err.println("Some Error Occured During Listing File");
-            }
-        } catch (RemoteException | InvalidProtocolBufferException ex) {
-            Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
-        }
-        
+    public ClientOperation(OperationType op, INameNode nn, String inputFilename, String outputFilename) {
+        this(op, nn);
+        this.inputFilename = inputFilename;
+        this.outputFilename = outputFilename;
     }
 
     public Integer openFileForWrite(String filename) {
@@ -111,37 +67,12 @@ public Integer openFileForWrite(String filename) {
         return handle;
     }
 
-    public List<Hdfs.BlockLocations> openFileForRead(String fileName) {
-        byte[] openRequest = ProtoMessage.openFileRequest(fileName, Boolean.TRUE);
-        List<Hdfs.BlockLocations> blockLocations = null;
-        try {
-            byte[] res = nn.openFile(openRequest);
-            Hdfs.OpenFileResponse openResponse = Hdfs.OpenFileResponse.parseFrom(res);
-            if(openResponse.getStatus() == 1) {
-                byte[] blockLocationRequest = ProtoMessage.blockLocationRequest(openResponse.getBlockNumsList());
-                byte[] blockLocationResponse = nn.getBlockLocations(blockLocationRequest);
-                Hdfs.BlockLocationResponse response = Hdfs.BlockLocationResponse.parseFrom(blockLocationResponse);
-                blockLocations = response.getBlockLocationsList();
-            }
-        } catch (RemoteException | InvalidProtocolBufferException ex) {
-            Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
-        }
-        return blockLocations;
-    }
-    
-    public void closeFile(Integer handle) {
-        byte[] closeRequest = ProtoMessage.closeFileRequest(handle);
-        try {
-            nn.closeFile(closeRequest);
-        } catch (Exception e) {}
-    }
-    
     IDataNode getDataNode(String ip, Integer port) {
         IDataNode  dn = null;
         try {
             dn = (IDataNode) Naming.lookup("rmi://" + ip
                     + ":" + port
-                    + "/" + DN_PREFIX);
+                    + "/" + Client.DN_PREFIX);
         } catch (NotBoundException | MalformedURLException | RemoteException ex) {
             Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
         }
@@ -152,14 +83,14 @@ public void putFile(String inputFileName, String outputFileName) {
 
         try {
             FileInputStream is = new FileInputStream(inputFileName);
-            byte chunk_data[] = new byte[CHUNK_SIZE];
+            byte chunk_data[] = new byte[Client.CHUNK_SIZE];
             Integer handle = openFileForWrite(outputFileName);
             byte[] assignBlockRequest = ProtoMessage.assignBlockRequest(handle);
             int read, chunk_number = 0;
 
             while ((read = is.read(chunk_data)) != -1) {
                 chunk_number++;
-                log("Chunk number: " + chunk_number);
+                Client.log("Chunk number: " + chunk_number);
                 Hdfs.BlockLocations blockLocations = null;
                 try {
                     byte[] response = nn.assignBlock(assignBlockRequest);
@@ -169,7 +100,7 @@ public void putFile(String inputFileName, String outputFileName) {
                 }
 
                 IDataNode dn = getDataNode(blockLocations.getLocations(0).getIp(), blockLocations.getLocations(0).getPort());
-                if(read != CHUNK_SIZE)   
+                if(read != Client.CHUNK_SIZE)   
                     chunk_data = Arrays.copyOfRange(chunk_data, 0, read);
 
                 byte[] writeBlockRequest = ProtoMessage.writeBlockRequest(chunk_data, blockLocations);
@@ -189,32 +120,57 @@ public void putFile(String inputFileName, String outputFileName) {
         }
     }
 
+    public void closeFile(Integer handle) {
+        byte[] closeRequest = ProtoMessage.closeFileRequest(handle);
+        try {
+            nn.closeFile(closeRequest);
+        } catch (Exception e) {}
+    }
+    
+    public List<Hdfs.BlockLocations> openFileForRead(String fileName) {
+        byte[] openRequest = ProtoMessage.openFileRequest(fileName, Boolean.TRUE);
+        List<Hdfs.BlockLocations> blockLocations = null;
+        try {
+            byte[] res = nn.openFile(openRequest);
+            Hdfs.OpenFileResponse openResponse = Hdfs.OpenFileResponse.parseFrom(res);
+            if(openResponse.getStatus() == 1) {
+                byte[] blockLocationRequest = ProtoMessage.blockLocationRequest(openResponse.getBlockNumsList());
+                byte[] blockLocationResponse = nn.getBlockLocations(blockLocationRequest);
+                Hdfs.BlockLocationResponse response = Hdfs.BlockLocationResponse.parseFrom(blockLocationResponse);
+                blockLocations = response.getBlockLocationsList();
+            }
+        } catch (RemoteException | InvalidProtocolBufferException ex) {
+            Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
+        }
+        return blockLocations;
+    }
+    
     public void getFile(String inputFileName, String outputFileName) {
         List<Hdfs.BlockLocations> blockLocations = openFileForRead(inputFileName);
         try {
             if(blockLocations != null) {
-                OutputStream os = new FileOutputStream(outputFileName);
-                for (Hdfs.BlockLocations block: blockLocations) {
-                    Random rand = new Random();
-                    Integer dataNodeInd = rand.nextInt(block.getLocationsCount());
-                    Hdfs.DataNodeLocation dnl = block.getLocations(dataNodeInd);
-                    log("Pulling block " + block.getBlockNumber() + " from DN " + dnl.getIp() + ":" + dnl.getPort());
-                    try {
-                        byte[] request = ProtoMessage.readBlockRequest(block.getBlockNumber());
-                        byte[] response = getDataNode(dnl.getIp(), dnl.getPort()).readBlock(request);
-                        Hdfs.ReadBlockResponse readBlockResponse = Hdfs.ReadBlockResponse.parseFrom(response);
-                        byte[] chunk_data = readBlockResponse.getData(0).toByteArray(); 
-                        os.write(chunk_data);
-                    } catch (RemoteException | InvalidProtocolBufferException ex) {
-                        Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
-                    } catch (IOException ex) {
-                        Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
-                    }
+                try (OutputStream os = new FileOutputStream(outputFileName)) {
+                    blockLocations.stream().forEach((block) -> {
+                        Random rand = new Random();
+                        Integer dataNodeInd = rand.nextInt(block.getLocationsCount());
+                        Hdfs.DataNodeLocation dnl = block.getLocations(dataNodeInd);
+                        Client.log("Pulling block " + block.getBlockNumber() + " from DN " + dnl.getIp() + ":" + dnl.getPort());
+                        try {
+                            byte[] request = ProtoMessage.readBlockRequest(block.getBlockNumber());
+                            byte[] response = getDataNode(dnl.getIp(), dnl.getPort()).readBlock(request);
+                            Hdfs.ReadBlockResponse readBlockResponse = Hdfs.ReadBlockResponse.parseFrom(response);
+                            byte[] chunk_data = readBlockResponse.getData(0).toByteArray();
+                            os.write(chunk_data);
+                        } catch (RemoteException | InvalidProtocolBufferException ex) {
+                            Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
+                        } catch (IOException ex) {
+                            Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
+                        }
+                    });
                 }
-                os.close();
                 System.out.println("File " + inputFileName + " pulled from mHdfs and stored on local system as " + outputFileName);
             } else {
-                System.out.println("File : " + inputFileName + " does not exist.");
+                System.out.println("File: " + inputFileName + " does not exist.");
             }
         } catch (FileNotFoundException ex) {
             Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
@@ -223,25 +179,120 @@ public void getFile(String inputFileName, String outputFileName) {
         }
     }
 
+    public void listFiles() {
+        byte[] inp = "*".getBytes();
+        try {
+            byte[] res = nn.list(inp);
+            Hdfs.ListFilesResponse listFilesResponse = Hdfs.ListFilesResponse.parseFrom(res);
+            if(listFilesResponse.getStatus() == 1) {
+                listFilesResponse.getFileNamesList().stream().forEach((fileName) -> {
+                    System.out.println(fileName);
+                });
+                System.out.println("");
+            } else {
+                System.err.println("Some Error Occured During Listing File");
+            }
+        } catch (RemoteException | InvalidProtocolBufferException ex) {
+            Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
+        }
+        
+    }
+    
+    @Override
+    public void run() {
+        switch(op) {
+            case PUT:
+                putFile(inputFilename, outputFilename);
+                break;
+            case GET:
+                getFile(inputFilename, outputFilename);
+                break;
+            case LIST:
+                listFiles();
+                break;
+        }
+    }
+    
+}
+
+public class Client {
+    static final String NN_NAME = "NameNode";
+    static final String DN_PREFIX = "DataNode";
+    static final Integer CHUNK_SIZE = 128*1024*1024;
+    
+    private static final Properties PROPS = new Properties();
+    
+    private INameNode nn = null;
+    
+    public static void log(String s) {
+        String op = String.valueOf(System.currentTimeMillis()) + " ";
+        op += "[Client] ";
+        op += ": ";
+        System.out.println(op + s);
+    }
+    
+    public static void main(String args[]) {
+        try {
+            PROPS.load(new BufferedReader(new FileReader("config.properties")));
+        } catch (IOException ex) {
+            Logger.getLogger(NameNode.class.getName()).log(Level.SEVERE, null, ex);
+        }
+        
+        Client client = new Client();
+        client.findnn();
+        client.mainloop();
+    }
+    
+    public void findnn() {
+        while(nn == null)
+        {
+            try {
+                nn = (INameNode) Naming.lookup("rmi://" + PROPS.getProperty("rmi.namenode.ip")
+                        + ":" + PROPS.getProperty("rmi.namenode.port") + "/" + NN_NAME);
+                log("Found Name Node");
+            } catch (NotBoundException | MalformedURLException | RemoteException ex) {
+                //Logger.getLogger(Client.class.getName()).log(Level.SEVERE, null, ex);
+            }
+            if (nn == null)
+                try {
+                    Thread.sleep(1000);
+                } catch (Exception e) {}
+        }
+    }
+    
     public void mainloop() {
+        ClientOperation clientOperation = null;
         Scanner in = new Scanner(System.in);
         for(;;) {
+            System.out.print("Command -> ");
             String line = in.nextLine();
             String[] ip = line.split(" ");
-            switch (ip[0]) {
+            Boolean quit = false;
+            switch (ip[0].toLowerCase()) {
                 case "put":
-                    putFile(ip[1], ip[2]);
+                    clientOperation = new ClientOperation(ClientOperation.OperationType.PUT, nn, ip[1], ip[2]);
                     break;
                 case "get":
-                    getFile(ip[1], ip[2]);
+                    clientOperation = new ClientOperation(ClientOperation.OperationType.GET, nn, ip[1], ip[2]);
                     break;
                 case "list":
-                    listFiles();
+                    clientOperation = new ClientOperation(ClientOperation.OperationType.LIST, nn);
                     break;
-                
-                default:
+                case "exit":
+                    quit = true;
                     break;
+                    
+                default:
+                    System.out.println("Command " + ip[0] + " not recognized!");
+                    continue;
             }
+            if(quit)
+                break;
+            Thread t = new Thread(clientOperation);
+            t.start();
         }
+        System.out.println("Thank you for using mDFS!");
+        System.out.println("Please hit Ctrl+C to kill all related services.");
+        for(;;) {}
     }
 }