//import java.net.*;
//import java.io.*;
import java.util.Date;
import java.util.Vector;
import java.util.HashMap;
import java.io.IOException;
import java.util.StringTokenizer;
import java.io.File;
import java.nio.channels.FileChannel;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.io.PrintWriter;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.io.FileNotFoundException;

public class JobQServer{

  private class Job {
    private int[] taskStatus=null;
    private int progress=-1;
    private int job_status=0;
    private static final int STATUS_WROTE=-1;
    private int[] jobSize=null;
    private int[] cumProd=null;
    private int allProd=0;
    
    private void init() {
      int k=jobSize.length;
      cumProd=new int[k];
      cumProd[k-1]=1;
      for (int i=k-1; i>0; i--) {
	cumProd[i-1]=cumProd[i]*jobSize[i];
      }
      allProd=cumProd[0]*jobSize[0];
      taskStatus=new int[allProd];
      for (int i=0; i<taskStatus.length; i++) {
	taskStatus[i]=0;
      }
      job_status=0;
      //      System.err.println("allProd:"+allProd);
    }

    private int[] ind2sub (int ind) throws ArrayIndexOutOfBoundsException{
      int[] retVal=new int[cumProd.length];
      int curr=ind-1;
      if (curr < 0) {
	for (int i=0; i<cumProd.length; i++) {
	  retVal[i]=0;
	}
      } else if (curr>allProd-1) {
	throw new ArrayIndexOutOfBoundsException();
      } else {
	for (int i=0; i<cumProd.length; i++) {
	  retVal[i]= curr / cumProd[i] + 1;
	  curr = curr % cumProd[i];
	}
      }
      return retVal;
    }

    private int sub2ind(int[] sub) throws ArrayIndexOutOfBoundsException{
      int retVal=0;
      for (int i=0; i<cumProd.length; i++) {
	if (sub[i]-1 > jobSize[i]-1) {
	  throw new ArrayIndexOutOfBoundsException();
	} else {
	  retVal+=(sub[i]-1)*cumProd[i];
	}
      }
      retVal++;
      return retVal;
    }

    public Job(int[] js) {
      jobSize=new int[js.length];
      for (int i=0; i<js.length; i++){
	jobSize[i]=js[i];
	//	System.err.println("jobSize[i]="+jobSize[i]);
      }
      init();
    }
    public Job(StringTokenizer js) {
      int k=js.countTokens();
      jobSize=new int[k];
      for (int i=0; i<k; i++){
	jobSize[i]=Integer.parseInt(js.nextToken());
	//	System.err.println("jobSize[i]="+jobSize[i]);
      }
      init();
    }

    public int[] next() {
      if (job_status!=0) {
	return ind2sub(0);
      }
      int nextInd=0;
      if (progress < allProd-1) {
	// Not all tasks have been touched
	progress++;
	taskStatus[progress]=1;
	nextInd=progress;
      } else {
	// Grab the task with the least number of slaves
	int minN=Integer.MAX_VALUE, minI=-1;
	for (int i=0; i<allProd; i++) {
	  if (taskStatus[i]!=STATUS_WROTE && taskStatus[i]<minN) {
	    minN=taskStatus[i];
	    minI=i;
	  }
	}
	if (minI==-1) {
	  // All tasks are done
	  nextInd=-1;
	} else{
	  taskStatus[minI]++;
	  nextInd=minI;
	}
      }
      return ind2sub(nextInd+1);
    }
    public String nextStr() {
      int[] vi=next();
      String retVal="";
      for (int i=0; i<vi.length; i++) {
	retVal=retVal+vi[i]+" ";
      }
      return retVal;
    }
    
    private class CopyFileThread extends Thread {
      FileChannel in=null;
      FileChannel out=null;
      public CopyFileThread(String sf, String st) throws Exception{
	in=new FileInputStream(new File(sf)).getChannel();
	out=new  FileOutputStream(new File(st)).getChannel();
      }
      public void run() {
	try {
	  in.transferTo( 0, in.size(), out);
	} catch (FileNotFoundException e) {
	  System.err.println(e);
	} catch (IOException e) {
	  System.err.println(e);
	}		  
      }
    }
	
    public final static int WROTE_FINE=0;
    public final static int WROTE_NOT_ASSIGNED=-1;
    public final static int WROTE_ALREADY_WRITTEN=-2;
    public final static int WROTE_FILECOPY_ERROR=-3;
    public int wrote(int[] vi, String tmpFile, String destFile) {
      int ind;
      try {
       ind=sub2ind(vi)-1;
      } catch (ArrayIndexOutOfBoundsException e) {
	job_status=-1;
	return WROTE_NOT_ASSIGNED;
      }	
      if (taskStatus[ind]==0) {
	job_status=-1;
	return WROTE_NOT_ASSIGNED;
      }
      if (taskStatus[ind]==STATUS_WROTE) {
	return WROTE_ALREADY_WRITTEN;
      } else {
	// if NOT already written
	// start a thread to copy over files
	try {
	  new CopyFileThread(tmpFile,destFile).start();
	  taskStatus[ind]=STATUS_WROTE;
	  return WROTE_FINE;
	} catch (Exception e) {
	  job_status=-1;
	  return WROTE_FILECOPY_ERROR;
	}
      }
    }

    public void reset() {
      for (int i=0; i<taskStatus.length; i++) {
	taskStatus[i]=0;
      }
      progress=-1;
      job_status=0;
    }

    public void clear() {
      taskStatus=null;
      jobSize=null;
      cumProd=null;
      progress=-1;
      job_status=0;
    }

    public int jobNumDim() {
      return jobSize.length;
    }

    public int getProgress() {
      return progress;
    }
    
    public int getNumDone() {
      int retVal=0;
      for (int i=0; i<taskStatus.length; i++) {
	if (taskStatus[i]==STATUS_WROTE){
	  retVal++;
	}
      }
      return retVal;
    }
  }




  private static  String vi2str(int[] vi) {
    if (vi.length==0) {
      return "";
    } else {
      String retVal="(";
      for (int i=0; i<vi.length-1; i++) {
	retVal=retVal+vi[i]+",";
      }
      retVal=retVal+vi[vi.length-1]+")";
      return retVal;
    }
  }




  public final static int JOBQSERVER_PORT=30000;

  private HashMap<String, Job> job=null;
  private ServerSocket serverSocket=null;
  public JobQServer() {
    job=new HashMap<String, Job>();
    // Start server 
    int port=JOBQSERVER_PORT;
    try {
      serverSocket = new ServerSocket(port);
    } catch (IOException e) {
      System.err.println("Could not listen on port: "+port+".");
      System.exit(1);
    }
    System.err.println("JobQServer running..");
  }

  public void listenRespondLoop() throws IOException {
    // Loop over client connections
    Socket clientSocket = null;
    while (true) {
      try {
	clientSocket = serverSocket.accept();
      } catch (IOException e) {
	System.err.println("Accept failed.");
	System.exit(1);
      }

      // set up input/output with the client
      PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true);
      BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
      String outStr=null;
      String data=in.readLine();

      try {
	// write initial server log
	String dateStr=new SimpleDateFormat("MMM d yyyy HH:mm:ss Z").format(new java.util.Date()); 
	String clientUrlStr=clientSocket.getInetAddress().getHostName();
	System.err.print("--DATE-- "+dateStr+" --FROM-- "+clientUrlStr+" --DATA-- "+data);
	
	// execute client request
	StringTokenizer st=new StringTokenizer(data);
	String arg1=st.nextToken();
	if (arg1.equals("HAOexit")) {
	  System.exit(0);
	} else if (arg1.equals("clear")) {
	  job.clear();
	} else {
	  String jobname=arg1;
	  String arg2=st.nextToken();
	  if (arg2.equals("clear")) {
	    if (job.containsKey(jobname)) {
	      job.get(jobname).clear();
	      outStr="job "+jobname+" cleared"; 
	    } else {
	      outStr="job "+jobname+" cleared (didn't exist)"; 
	    }
	  } else if (arg2.equals("reset")) {
	    job.get(jobname).reset();
	    outStr="job "+jobname+" reset";
	  } else if (arg2.equals("next")) {
	    if (job.containsKey(jobname)) {
	      outStr=job.get(jobname).nextStr();
	    } else {
	      outStr="job "+jobname+" not allocated yet.";
	    }
	  } else if (arg2.equals("alloc")) {
	    // 	  Vector<Integer> jobSize=new Vector();
	    // 	  while (st.hasMoreTokens()) {
	    // 	    jobSize.add(Integer.parseInt(st.nextToken()));
	    // 	  }
	    if (job.containsKey(jobname)) {
	      // help out on garbage collection 
	      job.remove(jobname);
	    }
	    job.put(jobname,new Job(st));
	    outStr="job "+jobname+" allocated";
	  } else if (arg2.equals("wrote")) {
	    int k=job.get(jobname).jobNumDim();
	    int[] vi=new int[k];
	    for (int i=0; i<k; i++) {
	      vi[i]=Integer.parseInt(st.nextToken());
	    }
	    String fromFile=st.nextToken();
	    String toFile=st.nextToken();
	    int writing_status=job.get(jobname).wrote(vi, fromFile, toFile);
	    if (writing_status==Job.WROTE_FINE) {
	      outStr="job "+jobname+ ": server ack task "
		+vi2str(vi)+" has been done"; 
	    } else if (writing_status==Job.WROTE_ALREADY_WRITTEN) {
	      outStr="job "+jobname+ ": task "
		+vi2str(vi)+" has already been written";
	    } else if (writing_status==Job.WROTE_NOT_ASSIGNED) {
	      outStr="job "+jobname+ ": task "
		+vi2str(vi)+" has not been assigned!"; 
	    } else if (writing_status==Job.WROTE_FILECOPY_ERROR) {
	      outStr="job "+jobname+": task "
		+vi2str(vi)+" has file copy error!";
	    } else {
	      System.err.println("Should NOT be in this branch!");
	    }
	  } else {
	    System.err.println("Unknown client request arg2:"+arg2);
	  }
	  // write server status to server log
	  System.err.print(" --STATUS-- #jobs="+job.size());
	  if (job.containsKey(jobname))
	    System.err.print(" #tasks in "+jobname+" assigned="+(job.get(jobname).getProgress()+1)+" done="+job.get(jobname).getNumDone());
	}

	// write response to server log
	System.err.println(" ---> --REPLY-- "+outStr);

	// write to client
	out.println(outStr); out.flush(); 
      } catch (Exception e) {
	System.err.println("Error during processing client command.");
	e.printStackTrace();
      }
	
      out.close();
      in.close();
      clientSocket.close();
    }
    //    serverSocket.close();
  }

  public static void main(String[] args) throws IOException {
    JobQServer jqserver=new JobQServer();
    jqserver.listenRespondLoop();
  }
}