jpayne@68: package fileIO;
jpayne@68: import java.io.InputStream;
jpayne@68: import java.util.Arrays;
jpayne@68: import java.util.concurrent.ArrayBlockingQueue;
jpayne@68: 
jpayne@68: import shared.Timer;
jpayne@68: import shared.Tools;
jpayne@68: 
jpayne@68: 
jpayne@68: /**
jpayne@68:  * Runs a ByteFile1 in a separate thread.  Can speed up disk reading, particularly of compressed files, at cost of slightly more work done.
jpayne@68:  * Drop-in compatible with ByteFile1.
jpayne@68:  * @author Brian Bushnell
jpayne@68:  * @date Sep 23, 2013
jpayne@68:  *
jpayne@68:  */
jpayne@68: public final class ByteFile2 extends ByteFile {
jpayne@68: 	
jpayne@68: 	
jpayne@68: 	public static void main(String[] args){
jpayne@68: 		ByteFile2 tf=new ByteFile2(args.length>0 ? args[0] : "stdin", true);
jpayne@68: 		long first=0, last=100;
jpayne@68: 		boolean speedtest=false;
jpayne@68: 		if(args.length>1){
jpayne@68: 			if(args[1].equalsIgnoreCase("speedtest")){
jpayne@68: 				speedtest=true;
jpayne@68: 				first=0;
jpayne@68: 				last=Long.MAX_VALUE;
jpayne@68: 			}else{
jpayne@68: 				first=Integer.parseInt(args[1]);
jpayne@68: 				last=first+100;
jpayne@68: 			}
jpayne@68: 		}
jpayne@68: 		if(args.length>2){
jpayne@68: 			last=Integer.parseInt(args[2]);
jpayne@68: 		}
jpayne@68: 		speedtest(tf, first, last, !speedtest);
jpayne@68: 		
jpayne@68: 		tf.close();
jpayne@68: 		tf.reset();
jpayne@68: 		tf.close();
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	private static void speedtest(ByteFile2 tf, long first, long last, boolean reprint){
jpayne@68: 		Timer t=new Timer();
jpayne@68: 		long lines=0;
jpayne@68: 		long bytes=0;
jpayne@68: 		for(long i=0; i<first; i++){tf.nextLine();}
jpayne@68: 		if(reprint){
jpayne@68: 			for(long i=first; i<last; i++){
jpayne@68: 				byte[] s=tf.nextLine();
jpayne@68: 				if(s==null){break;}
jpayne@68: 
jpayne@68: 				lines++;
jpayne@68: 				bytes+=s.length;
jpayne@68: 				System.out.println(new String(s));
jpayne@68: 			}
jpayne@68: 			
jpayne@68: 			System.err.println("\n");
jpayne@68: 			System.err.println("Lines: "+lines);
jpayne@68: 			System.err.println("Bytes: "+bytes);
jpayne@68: 		}else{
jpayne@68: 			for(long i=first; i<last; i++){
jpayne@68: 				byte[] s=tf.nextLine();
jpayne@68: 				if(s==null){break;}
jpayne@68: 				lines++;
jpayne@68: 				bytes+=s.length;
jpayne@68: 			}
jpayne@68: 		}
jpayne@68: 		t.stop();
jpayne@68: 		
jpayne@68: 		if(!reprint){
jpayne@68: 			System.err.println(Tools.timeLinesBytesProcessed(t, lines, bytes, 8));
jpayne@68: 		}
jpayne@68: 	}
jpayne@68: 	
jpayne@68: //	public ByteFile2(String name()){this(name(), false);}
jpayne@68: 	
jpayne@68: 	public ByteFile2(String fname, boolean allowSubprocess_){
jpayne@68: 		this(FileFormat.testInput(fname, FileFormat.TEXT, null, allowSubprocess_, false));
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	public ByteFile2(FileFormat ff){
jpayne@68: 		super(ff);
jpayne@68: 		if(verbose){System.err.println("ByteFile2("+ff+")");}
jpayne@68: 		open();
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	@Override
jpayne@68: 	public final void reset(){
jpayne@68: 		close();
jpayne@68: 		open();
jpayne@68: 		superReset();
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	@Override
jpayne@68: 	public synchronized final boolean close(){
jpayne@68: 		if(verbose){System.err.println("ByteFile2("+name()+").close()");}
jpayne@68: 		if(isOpen()){
jpayne@68: //			errorState|=ReadWrite.killProcess(name());
jpayne@68: 			thread.shutdown();
jpayne@68: 			while(thread.getState()!=Thread.State.TERMINATED){
jpayne@68: 				try {
jpayne@68: 					thread.join();
jpayne@68: 				} catch (InterruptedException e) {
jpayne@68: 					// TODO Auto-generated catch block
jpayne@68: 					e.printStackTrace();
jpayne@68: 				}
jpayne@68: 			}
jpayne@68: 			thread.bf1.close();
jpayne@68: 		}
jpayne@68: 		thread=null;
jpayne@68: 		currentList=null;
jpayne@68: 		currentLoc=0;
jpayne@68: //		assert(numIn==numOut) : numIn+", "+numOut;
jpayne@68: 		pushBack=null;
jpayne@68: 		if(verbose){System.err.println("ByteFile2("+name()+").close() returned "+errorState);}
jpayne@68: 		return errorState;
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	@Override
jpayne@68: 	public final byte[] nextLine(){
jpayne@68: 		
jpayne@68: 		if(pushBack!=null){//Commenting out does not seem to improve speed here.
jpayne@68: 			byte[] temp=pushBack;
jpayne@68: 			pushBack=null;
jpayne@68: 			return temp;
jpayne@68: 		}
jpayne@68: 		
jpayne@68: //		if(verbose){System.err.println("Reading line.");}
jpayne@68: //		byte[] r=null;
jpayne@68: 		
jpayne@68: 		byte[][] temp=currentList;
jpayne@68: 		int tempLoc=currentLoc;
jpayne@68: 		
jpayne@68: 		if(temp==null || tempLoc>=temp.length || temp[tempLoc]==null){
jpayne@68: 			boolean b=getBuffer();
jpayne@68: 			if(!b){
jpayne@68: 				if(verbose2){System.err.println("nextLine()->getBuffer() returned false.");}
jpayne@68: 				return null;
jpayne@68: 			}
jpayne@68: 			temp=currentList;
jpayne@68: 			tempLoc=currentLoc;
jpayne@68: 			if(temp==null || temp==poison || temp[tempLoc]==null){
jpayne@68: 				return null;
jpayne@68: 			}
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		//TODO: This is a race condition; currentList can be changed to null.  A defensive copy could be created.
jpayne@68: 		//Note that I read the above warning and added "temp" and "temploc" but I'm not sure if that fixed anything.
jpayne@68: 		assert(temp!=null && temp!=poison);
jpayne@68: 		assert(tempLoc<temp.length);
jpayne@68: 		assert(temp[tempLoc]!=null);
jpayne@68: 		byte[] r=temp[tempLoc];
jpayne@68: 		assert(r!=null);
jpayne@68: 		currentLoc++;
jpayne@68: //		numOut++;
jpayne@68: 		return r;
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	private boolean getBuffer(){
jpayne@68: 		if(verbose2){System.err.println("Getting new buffer.");}
jpayne@68: 		currentLoc=0;
jpayne@68: 		final BF1Thread bft=thread;
jpayne@68: 		if(bft==null){
jpayne@68: 			currentList=null;
jpayne@68: 			if(verbose2){System.err.println("No buffers available.  thread="+thread+", shutdown="+(thread==null ? "X" : ""+thread.shutdown));}
jpayne@68: 			return false;
jpayne@68: 		}
jpayne@68: 		if(currentList==poison){
jpayne@68: 			if(verbose2){System.err.println("A: Current list is poison.");}
jpayne@68: 			return false;
jpayne@68: 		}
jpayne@68: 		if(currentList!=null){
jpayne@68: 			Arrays.fill(currentList, null); //MUST be done or lines get recycled at end of file.
jpayne@68: 			while(currentList!=null){
jpayne@68: 				try {
jpayne@68: 					if(verbose2){System.err.println("adding to qEmpty list size "+currentList.length+"\n"+toString(currentList));}
jpayne@68: 					bft.qEmpty.put(currentList);
jpayne@68: 					currentList=null;
jpayne@68: 				} catch (InterruptedException e) {
jpayne@68: 					// TODO Auto-generated catch block
jpayne@68: 					e.printStackTrace();
jpayne@68: 				}
jpayne@68: 			}
jpayne@68: 		}
jpayne@68: 		assert(currentList==null);
jpayne@68: 		while(currentList==null){
jpayne@68: 			try {
jpayne@68: 				assert(bft!=null);
jpayne@68: 				if(verbose2){System.err.println("C: qFull.size()="+bft.qFull.size());}
jpayne@68: 				currentList=bft.qFull.take();
jpayne@68: 			} catch (InterruptedException e) {
jpayne@68: 				// TODO Auto-generated catch block
jpayne@68: 				e.printStackTrace();
jpayne@68: 			}
jpayne@68: 		}
jpayne@68: 		if(verbose2){
jpayne@68: 			if(currentList==poison){
jpayne@68: 				System.err.println("B: Current list is poison.");
jpayne@68: 			}else{
jpayne@68: 				System.err.println("getBuffer fetched a new buffer of size "+currentList.length);
jpayne@68: 			}
jpayne@68: 		}
jpayne@68: 		return currentList!=poison;
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	private final synchronized BF1Thread open(){
jpayne@68: 		if(verbose2){System.err.println("ByteFile2("+name()+").open()");}
jpayne@68: 		assert(thread==null);
jpayne@68: 		currentList=null;
jpayne@68: 		currentLoc=0;
jpayne@68: //		numIn=0;
jpayne@68: //		numOut=0;
jpayne@68: 		thread=new BF1Thread(ff);
jpayne@68: 		thread.start();
jpayne@68: 		return thread;
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	private class BF1Thread extends Thread{
jpayne@68: 		
jpayne@68: //		public BF1Thread(String fname){
jpayne@68: //			bf1=new ByteFile1(fname, false, allowSubprocess);
jpayne@68: //			qFull=new ArrayBlockingQueue<byte[][]>(buffs+2);
jpayne@68: //			qEmpty=new ArrayBlockingQueue<byte[][]>(buffs+2);
jpayne@68: //			for(int i=0; i<buffs; i++){
jpayne@68: //				try {
jpayne@68: //					qEmpty.put(new byte[bufflen][]);
jpayne@68: //				} catch (InterruptedException e) {
jpayne@68: //					// TODO Auto-generated catch block
jpayne@68: //					e.printStackTrace();
jpayne@68: //				}
jpayne@68: //			}
jpayne@68: //		}
jpayne@68: 		
jpayne@68: 		public BF1Thread(FileFormat ff){
jpayne@68: 			bf1=new ByteFile1(ff);
jpayne@68: 			qFull=new ArrayBlockingQueue<byte[][]>(buffs+2);
jpayne@68: 			qEmpty=new ArrayBlockingQueue<byte[][]>(buffs+2);
jpayne@68: 			for(int i=0; i<buffs; i++){
jpayne@68: 				try {
jpayne@68: 					qEmpty.put(new byte[bufflen][]);
jpayne@68: 				} catch (InterruptedException e) {
jpayne@68: 					// TODO Auto-generated catch block
jpayne@68: 					e.printStackTrace();
jpayne@68: 				}
jpayne@68: 			}
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		@Override
jpayne@68: 		public void run(){
jpayne@68: 			if(verbose){System.err.println("ByteFile2("+name()+").run()");}
jpayne@68: 			byte[] s=null;
jpayne@68: 			byte[][] list=null;
jpayne@68: 			while(list==null){
jpayne@68: 				try {
jpayne@68: 					list = qEmpty.take();
jpayne@68: 				} catch (InterruptedException e1) {
jpayne@68: 					// TODO Auto-generated catch block
jpayne@68: 					e1.printStackTrace();
jpayne@68: 				}
jpayne@68: 			}
jpayne@68: 			synchronized(this){
jpayne@68: 				if(list==poison || shutdown){
jpayne@68: 					shutdown();
jpayne@68: 					return;
jpayne@68: 				}
jpayne@68: 			}
jpayne@68: 			
jpayne@68: 			int loc=0;
jpayne@68: 			long bases=0;
jpayne@68: 			
jpayne@68: 			//At this point, list is not null
jpayne@68: 			for(s=bf1.nextLine(); s!=null; s=bf1.nextLine()){
jpayne@68: 				bases+=s.length;
jpayne@68: 				assert(list!=null) : "Somehow the list became null for "+bf1.name()+" at line "+cntr;
jpayne@68: 				list[loc]=s;
jpayne@68: 				loc++;
jpayne@68: //				numIn++;
jpayne@68: //				if(verbose){System.err.println("Added line "+numIn);}
jpayne@68: 				if(loc>=bufflen || bases>=buffcapacity){
jpayne@68: 					if(verbose2){System.err.println("Capacity exceeded.");}
jpayne@68: 					while(list!=null){
jpayne@68: 						try {
jpayne@68: //							synchronized(this){
jpayne@68: //								if(!shutdown){
jpayne@68: 									if(verbose2){
jpayne@68: 										System.err.println("A: Adding to qFull list of size "+loc);
jpayne@68: 										System.err.println(ByteFile2.toString(list));
jpayne@68: 									}
jpayne@68: 									cntr+=list.length;
jpayne@68: 									qFull.put(list);
jpayne@68: 									if(verbose2){System.err.println("A: qFull.size()="+qFull.size());}
jpayne@68: //								}
jpayne@68: //							}
jpayne@68: 							list=null;
jpayne@68: 							loc=0;
jpayne@68: 						} catch (InterruptedException e) {
jpayne@68: 							// TODO Auto-generated catch block
jpayne@68: 							e.printStackTrace();
jpayne@68: 						}
jpayne@68: 					}
jpayne@68: 					//At this point, list is null
jpayne@68: 					if(shutdown){
jpayne@68: 						if(verbose2){System.err.println("Break 1");}
jpayne@68: 						break;
jpayne@68: 					}
jpayne@68: 					while(list==null){
jpayne@68: 						if(verbose2){System.err.println("Taking empty list.");}
jpayne@68: 						try {
jpayne@68: 							list = qEmpty.take();
jpayne@68: 						} catch (InterruptedException e1) {
jpayne@68: 							// TODO Auto-generated catch block
jpayne@68: 							e1.printStackTrace();
jpayne@68: 						}
jpayne@68: 					}
jpayne@68: 					//At this point, list is not null
jpayne@68: 					bases=0;
jpayne@68: 					if(list==poison){
jpayne@68: 						if(verbose2){System.err.println("Break 2");}
jpayne@68: 						break;
jpayne@68: 					}
jpayne@68: 					//At this point, list is not null
jpayne@68: 				}
jpayne@68: 			}
jpayne@68: 			if(verbose2){System.err.println("Run loop exit.");}
jpayne@68: 			
jpayne@68: 			while(list!=null && loc>0){
jpayne@68: 				try {
jpayne@68: //					synchronized(this){
jpayne@68: //						if(!shutdown){
jpayne@68: 							if(verbose2){System.err.println("B: Adding list of size "+loc);}
jpayne@68: 							qFull.put(list);
jpayne@68: 							if(verbose2){System.err.println("B: qFull.size()="+qFull.size());}
jpayne@68: //						}
jpayne@68: //					}
jpayne@68: 					list=null;
jpayne@68: 					loc=0;
jpayne@68: 				} catch (InterruptedException e) {
jpayne@68: 					// TODO Auto-generated catch block
jpayne@68: 					e.printStackTrace();
jpayne@68: 				}
jpayne@68: 			}
jpayne@68: 			//At this point, list is null
jpayne@68: 			shutdown();
jpayne@68: 			
jpayne@68: 			if(verbose){System.err.println("ByteFile2("+name()+").run() finished");}
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		synchronized void shutdown(){
jpayne@68: 			if(verbose || verbose2){System.err.println("ByteFile2("+name()+").shutdown()");}
jpayne@68: 			if(shutdown){return;}
jpayne@68: 			shutdown=true;
jpayne@68: 			if(verbose2){System.err.println("Adding poison.");}
jpayne@68: 			qFull.add(poison);
jpayne@68: 			qEmpty.add(poison);
jpayne@68: 			if(verbose2){System.err.println("D: qFull.size()="+qFull.size());}
jpayne@68: 			if(verbose || verbose2){System.err.println("ByteFile2("+name()+").shutdown() finished");}
jpayne@68: 		}
jpayne@68: 		
jpayne@68: 		private boolean shutdown=false;
jpayne@68: 		final ByteFile1 bf1;
jpayne@68: 		final ArrayBlockingQueue<byte[][]> qFull;
jpayne@68: 		final ArrayBlockingQueue<byte[][]> qEmpty;
jpayne@68: 		
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	@Override
jpayne@68: 	public boolean isOpen(){
jpayne@68: 		final byte[][] list=currentList;
jpayne@68: 		final int loc=currentLoc;
jpayne@68: 		if(list!=null && loc<list.length && list[loc]!=null){return true;}
jpayne@68: 		final BF1Thread bft=thread;
jpayne@68: 		if(bft==null){
jpayne@68: 			return false;
jpayne@68: 		}
jpayne@68: 		return true;
jpayne@68: //		synchronized(bft){
jpayne@68: //		//NOTE!!!  This cannot be used because qFull.size() will not return a correctly synchronized value.  Poll() may work.
jpayne@68: //			assert(bft.bf1.isOpen() || !bft.qFull.isEmpty()) : bft.bf1.isOpen()+", "+bft.qFull.isEmpty()+", "+bft.qFull.size();
jpayne@68: //			return (bft.bf1.isOpen() || !bft.qFull.isEmpty());
jpayne@68: //		}
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	@Override
jpayne@68: 	public final void pushBack(byte[] line){
jpayne@68: 		assert(pushBack==null);
jpayne@68: 		pushBack=line;
jpayne@68: 	}
jpayne@68: 
jpayne@68: //	@Override
jpayne@68: //	public void pushBack(byte[] line) {
jpayne@68: //		if(bstart>line.length){
jpayne@68: //			bstart--;
jpayne@68: //			buffer[bstart]='\n';
jpayne@68: //			for(int i=0, j=bstart-line.length; i<line.length; i++, j++){
jpayne@68: //				buffer[j]=line[i];
jpayne@68: //			}
jpayne@68: //			bstart=bstart-line.length;
jpayne@68: //			return;
jpayne@68: //		}
jpayne@68: //		
jpayne@68: //		int bLen=bstop-bstart;
jpayne@68: //		int newLen=bLen+line.length+1;
jpayne@68: //		int rShift=line.length+1-bstart;
jpayne@68: //		assert(rShift>0) : bstop+", "+bstart+", "+line.length;
jpayne@68: //		while(newLen>buffer.length){
jpayne@68: //			//This could get big if pushback is used often,
jpayne@68: //			//unless special steps are taken to prevent it, like leaving extra space for pushbacks.
jpayne@68: //			buffer=Arrays.copyOf(buffer, buffer.length*2);
jpayne@68: //		}
jpayne@68: //		
jpayne@68: //		Tools.shiftRight(buffer, rShift);
jpayne@68: //		
jpayne@68: //		for(int i=0; i<line.length; i++){
jpayne@68: //			buffer[i]=line[i];
jpayne@68: //		}
jpayne@68: //		buffer[line.length]='\n';
jpayne@68: //		bstart=0;
jpayne@68: //		bstop=newLen;
jpayne@68: //	}
jpayne@68: 	
jpayne@68: 	/** For debugging */
jpayne@68: 	private static String toString(byte[][] x){
jpayne@68: 		StringBuilder sb=new StringBuilder();
jpayne@68: 		for(byte[] z : x){
jpayne@68: 			sb.append(z==null ? "null" : new String(z)).append('\n');
jpayne@68: 		}
jpayne@68: 		return sb.toString();
jpayne@68: 	}
jpayne@68: 	
jpayne@68: 	@Override
jpayne@68: 	public final InputStream is(){return thread==null ? null : thread.bf1.is();}
jpayne@68: 	
jpayne@68: 	@Override
jpayne@68: 	public final long lineNum(){return thread==null ? -1 : thread.bf1.lineNum();}
jpayne@68: 
jpayne@68: 	long cntr;
jpayne@68: 	private BF1Thread thread=null;
jpayne@68: 	private byte[][] currentList=null;
jpayne@68: 	private int currentLoc=0;
jpayne@68: //	private int currentSize=0;
jpayne@68: 	
jpayne@68: //	private long numIn=0, numOut=0;
jpayne@68: 	
jpayne@68: 	private byte[] pushBack=null;
jpayne@68: 	
jpayne@68: 	static final byte[][] poison=new byte[0][];
jpayne@68: 	public static boolean verbose=false;
jpayne@68: 	private static final boolean verbose2=false;
jpayne@68: 	private static final int bufflen=1000;
jpayne@68: 	private static final int buffs=4;
jpayne@68: 	private static final int buffcapacity=256000;
jpayne@68: 	
jpayne@68: 	private boolean errorState=false;
jpayne@68: 	
jpayne@68: }