<commit_before>/**
Copyright 2015 Acacia Team

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
 */

package org.acacia.server;

import java.rmi.UnknownHostException;
import x10.core.Thread;
import java.net.Socket;
import java.io.OutputStream;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.File;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import org.acacia.log.java.Logger_Java;
import org.acacia.server.java.AcaciaInstanceProtocol;
import org.acacia.server.java.AcaciaBackEndProtocol;
import java.io.PrintWriter;
import java.lang.InterruptedException;
import org.acacia.util.java.Conts_Java;
import org.acacia.util.java.Utils_Java;
import org.apache.commons.io.FileUtils;
import x10.util.HashMap;
import java.util.TreeSet;
import org.acacia.centralstore.AcaciaHashMapCentralStore;
import org.acacia.localstore.AcaciaLocalStore;
import org.acacia.localstore.AcaciaLocalStoreFactory;
import x10.util.HashSet;
import java.util.TreeMap;
import x10.util.StringBuilder;
import x10.util.ArrayList;

public class AcaciaManager {

 	public static def run() throws java.net.UnknownHostException{

 	}
 
 	public static def sendFileThroughService(val host:String, val port:Int, val fileName:String, val filePath:String){
 
 		// var t:Thread = new Thread(){
  	// 		public def run(){
  				try{
	  				val socket:Socket = new Socket(host,port);
	  				val out:OutputStream = socket.getOutputStream();
	  				val reader:BufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
	  				val file:File = new File(filePath);
	   				val bin:BufferedInputStream = new BufferedInputStream(new FileInputStream(file));
	   				out.write(x10.interop.Java.convert(fileName.bytes()));
	   				out.flush();
	   				val response:String = reader.readLine();
	   				if (response.equals(AcaciaInstanceProtocol.SEND_FILE)){
	   					Console.OUT.println("Sending file");
	   					val buffer:Rail[Byte] = new  Rail[byte](8*1024);
	   					var c:Int = -1n;
	
	   					while((c=bin.read(x10.interop.Java.convert[Byte](buffer)))>0){
	   						out.write(x10.interop.Java.convert[Byte](buffer), 0n, c);
	   						out.flush();
	   					}
   					}
  				}catch(e1:java.net.UnknownHostException){
                    e1.printStackTrace();
  				}catch(e2:java.io.IOException){
                    e2.printStackTrace();
  				}
  	// 		}
 		// };
 		// t.start();
 	}
 
	 /**
	  * The following communication happens with the designated AcaciaInstance which is identified the host and the port.
	  * 
	  * @param host
	  * @param port
	  * @param graphID
	  * @param filePath
	  * @param dataPort
	  * @return
	  */
 	public static def batchUploadFile(host:String, port:Int, graphID:long, filePath:String, dataPort:Int):boolean{
 		var result:boolean = true;
 
  		try{
  			val socket:Socket  = new Socket(host, port);
  			val out:PrintWriter = new PrintWriter(socket.getOutputStream());
  			val reader:BufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
  			var response:String = "";
 
 			//First we need to Handshake
 			out.println(AcaciaInstanceProtocol.HANDSHAKE);
 			out.flush();
 			response = reader.readLine();
 			 
 			if((response != null)&&(response.equals(AcaciaInstanceProtocol.HANDSHAKE_OK))){
  				out.println(java.net.InetAddress.getLocalHost().getHostName());
  				out.flush();
  			}
  
 		 	out.println(AcaciaInstanceProtocol.BATCH_UPLOAD);
 		 	out.flush();
 		 
		 	response = reader.readLine();

  			if((response != null)&&(response.equals(AcaciaInstanceProtocol.OK))){
			 	out.println(graphID);
			 	out.flush();
			 	response = reader.readLine();
			 
			 	val fileName:String = filePath.substring(filePath.lastIndexOf("/")+1n);
			 
				Console.OUT.println("filePath : " + filePath);
			 	Console.OUT.println("fileName : " + fileName);
			 
 				if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_NAME))){
 					out.println(fileName);
 					out.flush();
 
 					response = reader.readLine();
 
 					if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_LEN))){
 						Console.OUT.println("File len : " + new File(filePath).length());
 						out.println(new File(filePath).length());//File length in bytes
 						out.flush();
 
 						response = reader.readLine();
 						if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_CONT))){
 							sendFileThroughService(host, dataPort, fileName, filePath);
 						}							
  					}			
  				}
 
 				var counter:Int = 0n;
 
 				while(true){
 					out.println(AcaciaInstanceProtocol.FILE_RECV_CHK);
 					out.flush();
 					response = reader.readLine();

 					if((response != null)&&(response.equals(AcaciaInstanceProtocol.FILE_RECV_WAIT))){
 						Console.OUT.println("Checking file status : " + counter);
 						counter++;
 						
 						Thread.currentThread().sleep(1000);
 						//We sleep for 1 second, and try again.
 						continue;
 					}else{
 						break;
 					}
 				}
  				Console.OUT.println("File transfer complete...");
 
 				//Next we wait till the batch upload completes
 				while(true){
 					out.println(AcaciaInstanceProtocol.BATCH_UPLOAD_CHK);
 					out.flush();
 
 					response = reader.readLine();

  					if((response != null)&&(response.equals(AcaciaInstanceProtocol.BATCH_UPLOAD_WAIT))){
	 					Thread.currentThread().sleep(1000);
 						//We sleep for 1 second, and try again.
	 					continue;
	 				}else{
	 					break;
	 				}
  				}
 
	 			if((response != null)&&(response.equals(AcaciaInstanceProtocol.BATCH_UPLOAD_ACK))){
	 				Console.OUT.println("Batch upload completed...");	
	 			}else{
	 				Console.OUT.println("There was an error in the upload parocess.");
	 			}
	 		}
	 	}catch(e:java.net.UnknownHostException ){
	 		Logger_Java.error(e.getMessage());
	 		result = false;
	 	}catch(ec:java.io.IOException ){
	 		Logger_Java.error(ec.getMessage());
	 		result = false;
  		}	
 
 		return result;
 	}

 	public static def batchUploadCentralStore(host:String,port:Int,graphID:long,filePath:String,dataPort:int):boolean{
	 	var result:boolean = true;
	 
	 	Console.OUT.println(filePath);<commit_msg>Update AcaciaManager.x10<commit_after>/**
Copyright 2015 Acacia Team

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
 */

package org.acacia.server;

import java.rmi.UnknownHostException;
import x10.core.Thread;
import java.net.Socket;
import java.io.OutputStream;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.File;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import org.acacia.log.java.Logger_Java;
import org.acacia.server.java.AcaciaInstanceProtocol;
import org.acacia.server.java.AcaciaBackEndProtocol;
import java.io.PrintWriter;
import java.lang.InterruptedException;
import org.acacia.util.java.Conts_Java;
import org.acacia.util.java.Utils_Java;
import org.apache.commons.io.FileUtils;
import x10.util.HashMap;
import java.util.TreeSet;
import org.acacia.centralstore.AcaciaHashMapCentralStore;
import org.acacia.localstore.AcaciaLocalStore;
import org.acacia.localstore.AcaciaLocalStoreFactory;
import x10.util.HashSet;
import java.util.TreeMap;
import x10.util.StringBuilder;
import x10.util.ArrayList;

public class AcaciaManager {

 	public static def run() throws java.net.UnknownHostException{

 	}
 
 	public static def sendFileThroughService(val host:String, val port:Int, val fileName:String, val filePath:String){
 
 		// var t:Thread = new Thread(){
  	// 		public def run(){
  				try{
	  				val socket:Socket = new Socket(host,port);
	  				val out:OutputStream = socket.getOutputStream();
	  				val reader:BufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
	  				val file:File = new File(filePath);
	   				val bin:BufferedInputStream = new BufferedInputStream(new FileInputStream(file));
	   				out.write(x10.interop.Java.convert(fileName.bytes()));
	   				out.flush();
	   				val response:String = reader.readLine();
	   				if (response.equals(AcaciaInstanceProtocol.SEND_FILE)){
	   					Console.OUT.println("Sending file");
	   					val buffer:Rail[Byte] = new  Rail[byte](8*1024);
	   					var c:Int = -1n;
	
	   					while((c=bin.read(x10.interop.Java.convert[Byte](buffer)))>0){
	   						out.write(x10.interop.Java.convert[Byte](buffer), 0n, c);
	   						out.flush();
	   					}
   					}
  				}catch(e1:java.net.UnknownHostException){
                    e1.printStackTrace();
  				}catch(e2:java.io.IOException){
                    e2.printStackTrace();
  				}
  	// 		}
 		// };
 		// t.start();
 	}
 
	 /**
	  * The following communication happens with the designated AcaciaInstance which is identified the host and the port.
	  * 
	  * @param host
	  * @param port
	  * @param graphID
	  * @param filePath
	  * @param dataPort
	  * @return
	  */
 	public static def batchUploadFile(host:String, port:Int, graphID:long, filePath:String, dataPort:Int):boolean{
 		var result:boolean = true;
 
  		try{
  			val socket:Socket  = new Socket(host, port);
  			val out:PrintWriter = new PrintWriter(socket.getOutputStream());
  			val reader:BufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
  			var response:String = "";
 
 			//First we need to Handshake
 			out.println(AcaciaInstanceProtocol.HANDSHAKE);
 			out.flush();
 			response = reader.readLine();
 			 
 			if((response != null)&&(response.equals(AcaciaInstanceProtocol.HANDSHAKE_OK))){
  				out.println(java.net.InetAddress.getLocalHost().getHostName());
  				out.flush();
  			}
  
 		 	out.println(AcaciaInstanceProtocol.BATCH_UPLOAD);
 		 	out.flush();
 		 
		 	response = reader.readLine();

  			if((response != null)&&(response.equals(AcaciaInstanceProtocol.OK))){
			 	out.println(graphID);
			 	out.flush();
			 	response = reader.readLine();
			 
			 	val fileName:String = filePath.substring(filePath.lastIndexOf("/")+1n);
			 
				Console.OUT.println("filePath : " + filePath);
			 	Console.OUT.println("fileName : " + fileName);
			 
 				if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_NAME))){
 					out.println(fileName);
 					out.flush();
 
 					response = reader.readLine();
 
 					if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_LEN))){
 						Console.OUT.println("File len : " + new File(filePath).length());
 						out.println(new File(filePath).length());//File length in bytes
 						out.flush();
 
 						response = reader.readLine();
 						if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_CONT))){
 							sendFileThroughService(host, dataPort, fileName, filePath);
 						}							
  					}			
  				}
 
 				var counter:Int = 0n;
 
 				while(true){
 					out.println(AcaciaInstanceProtocol.FILE_RECV_CHK);
 					out.flush();
 					response = reader.readLine();

 					if((response != null)&&(response.equals(AcaciaInstanceProtocol.FILE_RECV_WAIT))){
 						Console.OUT.println("Checking file status : " + counter);
 						counter++;
 						
 						Thread.currentThread().sleep(1000);
 						//We sleep for 1 second, and try again.
 						continue;
 					}else{
 						break;
 					}
 				}
  				Console.OUT.println("File transfer complete...");
 
 				//Next we wait till the batch upload completes
 				while(true){
 					out.println(AcaciaInstanceProtocol.BATCH_UPLOAD_CHK);
 					out.flush();
 
 					response = reader.readLine();

  					if((response != null)&&(response.equals(AcaciaInstanceProtocol.BATCH_UPLOAD_WAIT))){
	 					Thread.currentThread().sleep(1000);
 						//We sleep for 1 second, and try again.
	 					continue;
	 				}else{
	 					break;
	 				}
  				}
 
	 			if((response != null)&&(response.equals(AcaciaInstanceProtocol.BATCH_UPLOAD_ACK))){
	 				Console.OUT.println("Batch upload completed...");	
	 			}else{
	 				Console.OUT.println("There was an error in the upload parocess.");
	 			}
	 		}
	 	}catch(e:java.net.UnknownHostException ){
	 		Logger_Java.error(e.getMessage());
	 		result = false;
	 	}catch(ec:java.io.IOException ){
	 		Logger_Java.error(ec.getMessage());
	 		result = false;
  		}	
 
 		return result;
 	}


	public static def batchUploadReplication(host:String, port:Int, graphID:long, filePath:String, dataPort:Int, placeID:Int):boolean{
 		var result:boolean = true;
 
  		try{
  			val socket:Socket  = new Socket(host, port);
  			val out:PrintWriter = new PrintWriter(socket.getOutputStream());
  			val reader:BufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
  			var response:String = "";
 
 			//First we need to Handshake
 			out.println(AcaciaInstanceProtocol.HANDSHAKE);
 			out.flush();
 			response = reader.readLine();
 			 
 			if((response != null)&&(response.equals(AcaciaInstanceProtocol.HANDSHAKE_OK))){
  				out.println(java.net.InetAddress.getLocalHost().getHostName());
  				out.flush();
  			}
  
 		 	out.println(AcaciaInstanceProtocol.BATCH_UPLOAD_REPLICATION);
 		 	out.flush();
 		 
		 	response = reader.readLine();

  			if((response != null)&&(response.equals(AcaciaInstanceProtocol.OK))){
				out.println(placeID);
				out.flush();
				response = reader.readLine();
				if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_GID))){				

				 	out.println(graphID);
				 	out.flush();
				 	response = reader.readLine();
				 
				 	val fileName:String = filePath.substring(filePath.lastIndexOf("/")+1n);
				 
					Console.OUT.println("filePath : " + filePath);
				 	Console.OUT.println("fileName : " + fileName);
				 
	 				if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_NAME))){
	 					out.println(fileName);
	 					out.flush();
	 
	 					response = reader.readLine();
	 
	 					if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_LEN))){
	 						Console.OUT.println("File len : " + new File(filePath).length());
	 						out.println(new File(filePath).length());//File length in bytes
	 						out.flush();
	 
	 						response = reader.readLine();
	 						if((response != null)&&(response.equals(AcaciaInstanceProtocol.SEND_FILE_CONT))){
	 							sendFileThroughService(host, dataPort, fileName, filePath);
	 						}							
	  					}			
	  				}
	 
	 				var counter:Int = 0n;
	 
	 				while(true){
	 					out.println(AcaciaInstanceProtocol.FILE_RECV_CHK);
	 					out.flush();
	 					response = reader.readLine();

	 					if((response != null)&&(response.equals(AcaciaInstanceProtocol.FILE_RECV_WAIT))){
	 						Console.OUT.println("Checking file status : " + counter);
	 						counter++;
	 						
	 						Thread.currentThread().sleep(1000);
	 						//We sleep for 1 second, and try again.
	 						continue;
	 					}else{
	 						break;
	 					}
	 				}
	  				Console.OUT.println("File transfer complete...");
	 
	 				//Next we wait till the batch upload completes
	 				while(true){
	 					out.println(AcaciaInstanceProtocol.BATCH_UPLOAD_CHK);
	 					out.flush();
	 
	 					response = reader.readLine();

	  					if((response != null)&&(response.equals(AcaciaInstanceProtocol.BATCH_UPLOAD_WAIT))){
		 					Thread.currentThread().sleep(1000);
	 						//We sleep for 1 second, and try again.
		 					continue;
		 				}else{
		 					break;
		 				}
	  				}
	 
		 			if((response != null)&&(response.equals(AcaciaInstanceProtocol.BATCH_UPLOAD_ACK))){
		 				Console.OUT.println("Batch upload completed...");	
		 			}else{
		 				Console.OUT.println("There was an error in the upload parocess.");
		 			}
				}
	 		}
	 	}catch(e:java.net.UnknownHostException ){
	 		Logger_Java.error(e.getMessage());
	 		result = false;
	 	}catch(ec:java.io.IOException ){
	 		Logger_Java.error(ec.getMessage());
	 		result = false;
  		}	
 
 		return result;
 	}


 	public static def batchUploadCentralStore(host:String,port:Int,graphID:long,filePath:String,dataPort:int):boolean{
	 	var result:boolean = true;
	 
	 	Console.OUT.println(filePath);
{'fraction_non_alphanumeric': 0.12675555555555557, 'fraction_numerical': 0.006518518518518518, 'mean_word_length': 3.180331929650731, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6801736', 'n_tokens_mistral': 6599, 'n_tokens_neox': 5347, 'n_words': 996}