package bloom; import java.util.Arrays; import java.util.Random; import java.util.concurrent.ArrayBlockingQueue; import shared.Shared; import shared.Timer; /** * * Uses hashing rather than direct-mapping to support longer kmers. * * @author Brian Bushnell * @date Aug 17, 2012 * */ public class KCountArray4MT extends KCountArray { /** * */ private static final long serialVersionUID = -575682515926973788L; public static void main(String[] args){ long cells=Long.parseLong(args[0]); int bits=Integer.parseInt(args[1]); int gap=Integer.parseInt(args[2]); int hashes=Integer.parseInt(args[3]); verbose=false; KCountArray4MT kca=new KCountArray4MT(cells, bits, gap, hashes); System.out.println(kca.read(0)); kca.increment(0); System.out.println(kca.read(0)); kca.increment(0); System.out.println(kca.read(0)); System.out.println(); System.out.println(kca.read(1)); kca.increment(1); System.out.println(kca.read(1)); kca.increment(1); System.out.println(kca.read(1)); System.out.println(); System.out.println(kca.read(100)); kca.increment(100); System.out.println(kca.read(100)); kca.increment(100); System.out.println(kca.read(100)); kca.increment(100); System.out.println(kca.read(100)); System.out.println(); System.out.println(kca.read(150)); kca.increment(150); System.out.println(kca.read(150)); System.out.println(); } public KCountArray4MT(long cells_, int bits_, int gap_, int hashes_){ super(cells_, bits_, gap_); // verbose=false; long words=cells/cellsPerWord; long x=(words/numArrays); if(x>=Integer.MAX_VALUE){x=Integer.MAX_VALUE-3;} assert(x<=Integer.MAX_VALUE); wordsPerArray=(int)(x); cellsPerArray=cells/numArrays; cellMod=cellsPerArray-1; hashes=hashes_; // System.out.println("cells="+cells+", words="+words+", wordsPerArray="+wordsPerArray+", numArrays="+numArrays+", hashes="+hashes); // assert(false); matrix=new int[numArrays][]; assert(hashes>0 && hashes<=hashMasks.length); } @Override public int read(final long rawKey){ assert(finished); if(verbose){System.err.println("Reading raw key "+rawKey);} long key2=hash(rawKey, 0); int min=readHashed(key2); for(int i=1; i0; i++){ if(verbose){System.err.println("Reading. i="+i+", key2="+key2);} key2=Long.rotateRight(key2, hashBits); key2=hash(key2, i); if(verbose){System.err.println("Rot/hash. i="+i+", key2="+key2);} min=min(min, readHashed(key2)); } return min; } private int readHashed(long key){ if(verbose){System.err.print("Reading hashed key "+key);} // System.out.println("key="+key); int arrayNum=(int)(key&arrayMask); key=(key>>>arrayBits)%(cellMod); // key=(key>>>(arrayBits+1))%(cellMod); // System.out.println("array="+arrayNum); // System.out.println("key2="+key); int[] array=matrix[arrayNum]; int index=(int)(key>>>indexShift); // assert(false) : indexShift; // System.out.println("index="+index); int word=array[index]; // System.out.println("word="+Integer.toHexString(word)); assert(word>>>(cellBits*key) == word>>>(cellBits*(key&cellMask))); // int cellShift=(int)(cellBits*(key&cellMask)); int cellShift=(int)(cellBits*key); if(verbose){System.err.println(", array="+arrayNum+", index="+index+", cellShift="+(cellShift%32)+", value="+((int)((word>>>cellShift)&valueMask)));} // System.out.println("cellShift="+cellShift); return (int)((word>>>cellShift)&valueMask); } @Override public void write(final long key, int value){ throw new RuntimeException("Not allowed for this class."); } // @Override // /** This should increase speed by doing the first hash outside the critical section, but it does not seem to help. */ // public void increment(long[] keys){ // for(int i=0; i>>=cellBits; comma=", "; } } } sb.append("]"); return sb.toString(); } @Override public double usedFraction(){return cellsUsed/(double)cells;} @Override public double usedFraction(int mindepth){return cellsUsed(mindepth)/(double)cells;} @Override public long cellsUsed(int mindepth){ long count=0; for(int[] array : matrix){ if(array!=null){ for(int word : array){ while(word>0){ int x=word&valueMask; if(x>=mindepth){count++;} word>>>=cellBits; } } } } return count; } @Override final long hash(long key, int row){ int cell=(int)((Long.MAX_VALUE&key)%(hashArrayLength-1)); // int cell=(int)(hashCellMask&(key)); if(row==0){//Doublehash only first time key=key^hashMasks[(row+4)%hashMasks.length][cell]; cell=(int)(hashCellMask&(key>>4)); // cell=(int)(hashCellMask&(key>>hashBits)); // cell=(int)((Long.MAX_VALUE&key)%(hashArrayLength-1)); } return key^hashMasks[row][cell]; } /** * @param i * @param j * @return */ private static long[][] makeMasks(int rows, int cols) { long seed; synchronized(KCountArray4MT.class){ seed=counter; counter++; } Timer t=new Timer(); long[][] r=new long[rows][cols]; Random randy=Shared.threadLocalRandom(seed); for(int i=0; i200000000L){System.out.println("Mask-creation time: "+t);} return r; } private static void fillMasks(long[] r, Random randy) { // for(int i=0; i16){ x&=(~(1L<16){ x&=(~(1L<<(randy.nextInt(32)+32))); } // System.out.print("."); // y=(((int)(x&mask))^i); y=(((int)(x&mask))); z=(int)((x>>hashBits)&mask); if(count1[y]>0 || count2[z]>0){ x=0; } } // System.out.println(Long.toBinaryString(x)); r[i]=(x&Long.MAX_VALUE); count1[y]++; count2[z]++; } } @Override public void initialize(){ for(int i=0; i0){ writers[i].add(array); } } //Add poison for(WriteThread wt : writers){ wt.add(poison); } //Wait for termination for(WriteThread wt : writers){ // System.out.println("wt"+wt.num+" is alive: "+wt.isAlive()); while(wt.isAlive()){ // System.out.println("wt"+wt.num+" is alive: "+wt.isAlive()); try { wt.join(10000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } if(wt.isAlive()){System.err.println(wt.getClass().getCanonicalName()+" is taking a long time to die.");} } cellsUsed+=wt.cellsUsedPersonal; // System.out.println("cellsUsed="+cellsUsed); } assert(!finished); finished=true; } } private class WriteThread extends Thread{ public WriteThread(int tnum){ num=tnum; } @Override public void run(){ assert(matrix[num]==null); array=new int[wordsPerArray]; //Makes NUMA systems use local memory. matrix[num]=array; long[] keys=null; while(!shutdown){ if(verbose){System.err.println(" - Reading keys for wt"+num+".");} while(keys==null){ try { keys=writeQueue.take(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } if(keys==poison){ // assert(false); shutdown=true; }else{ for(long key : keys){ incrementHashedLocal(key); } } // System.out.println(" -- Read keys for wt"+num+". poison="+(keys==poison)+", len="+keys.length); if(verbose){System.err.println(" -- Read keys for wt"+num+". (success)");} keys=null; if(verbose){System.err.println("shutdown="+shutdown);} } // System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I died: "+shutdown+", "+(keys==null)+"."); // assert(false) : ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I died: "+shutdown+", "+(keys==null)+"."; array=null; } void add(long[] keys){ // assert(isAlive()); assert(!shutdown); if(shutdown){return;} // assert(keys!=poison); if(verbose){System.err.println(" + Adding keys to wt"+num+".");} boolean success=false; while(!success){ try { writeQueue.put(keys); success=true; } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } if(verbose){System.err.println(" ++ Added keys to wt"+num+". (success)");} } private int incrementHashedLocal(long key){ assert((key&arrayMask)==num); key=(key>>>arrayBits)%(cellMod); // key=(key>>>(arrayBits+1))%(cellMod); int index=(int)(key>>>indexShift); int word=array[index]; int cellShift=(int)(cellBits*key); int value=((word>>>cellShift)&valueMask); if(value==0){cellsUsedPersonal++;} value=min(value+1, maxValue); word=(value< writeQueue=new ArrayBlockingQueue(16); public boolean shutdown=false; } public long cellsUsed(){return cellsUsed;} private boolean finished=false; private long cellsUsed; final int[][] matrix; private final WriteThread[] writers=new WriteThread[numArrays]; private final int hashes; final int wordsPerArray; private final long cellsPerArray; final long cellMod; private final long[][] hashMasks=makeMasks(8, hashArrayLength); private final long[][] buffers=new long[numArrays][1000]; private final int[] bufferlen=new int[numArrays]; private static final int hashBits=6; private static final int hashArrayLength=1<