Index: test/org/apache/lucene/store/TestBufferPoolLRU.java =================================================================== --- test/org/apache/lucene/store/TestBufferPoolLRU.java (revision 0) +++ test/org/apache/lucene/store/TestBufferPoolLRU.java (revision 0) @@ -0,0 +1,135 @@ +package org.apache.lucene.store; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Random; + +import junit.framework.TestCase; + +public class TestBufferPoolLRU extends TestCase { + + class FakeIndexInput extends BufferedIndexInput { + long length; + + FakeIndexInput(long length) { + this.length = length; + } + + protected void readInternal(byte[] b, int offset, int length) + throws IOException { + } + + protected void seekInternal(long pos) throws IOException { + } + + public void close() throws IOException { + } + + public long length() { + return length; + } + } + + class BufferPoolLRU2 extends BufferPoolLRU { + boolean longRead; + + BufferPoolLRU2(int bufferSize, int numBuffers, boolean longRead) { + super(bufferSize, numBuffers); + this.longRead = longRead; + } + + protected void readInput(BareInput input, long position, byte[] b, + int offset, int len) throws IOException { + for (int i = 0; i < len; i++) { + b[offset + i] = (byte) ((position + i) % Byte.MAX_VALUE); + } + try { + if (longRead) { + Thread.sleep(10); + } + } catch (Exception e) { + } + } + } + + class BufferPoolTest { + final int length; + final int bufferSize; + final FakeIndexInput input; + final BufferPool pool; + final ArrayList eList; + + BufferPoolTest(boolean longRead) { + length = 262189; + bufferSize = 8531; + int numBuffers = 19; + + input = new FakeIndexInput(length); + pool = new BufferPoolLRU2(bufferSize, numBuffers, longRead); + eList = new ArrayList(); + } + + void runTest(int numThreads, int numReads) throws Exception { + ReadThread[] threads = new ReadThread[numThreads]; + for (int i = 0; i < threads.length; i++) { + threads[i] = new ReadThread(i, numReads); + } + for (int i = 0; i < threads.length; i++) { + threads[i].start(); + } + for (int i = 0; i < threads.length; i++) { + threads[i].join(); + } + if (eList.size() > 0) { + throw (Exception)eList.get(0); // throw the first + } + } + + class ReadThread extends Thread { + String fakeFileName = new String("c.txt"); + Random r; + byte[] local; + int numReads; + + ReadThread(int seed, int numReads) { + this.r = new Random(seed); + this.local = new byte[r.nextInt(bufferSize) + 1]; + this.numReads = numReads; + } + + public void run() { + for (int i = 0; i < numReads; i++) { + int position = r.nextInt(length); + int offset = r.nextInt(local.length); + int len = r.nextInt(local.length - offset); + if (position + len > length) { + len = length - position; + } + + try { + pool.read(fakeFileName, position, local, offset, len, input); + } catch (IOException e) { + synchronized (eList) { + eList.add(e); + } + } + + for (int j = 0; j < len; j++) { + byte expected = (byte) ((position + j) % Byte.MAX_VALUE); + assertEquals(expected, local[offset + j]); + } + } + } + } + } + + public void testBufferPoolShort() throws Exception { + BufferPoolTest test = new BufferPoolTest(false); + test.runTest(20, 1000); + } + + public void testBufferPoolLong() throws Exception { + BufferPoolTest test = new BufferPoolTest(true); + test.runTest(20, 1000); + } +} Index: test/org/apache/lucene/store/_TestHelper.java =================================================================== --- test/org/apache/lucene/store/_TestHelper.java (revision 588053) +++ test/org/apache/lucene/store/_TestHelper.java (working copy) @@ -29,7 +29,7 @@ /** Returns true if the instance of the provided input stream is actually * an FSIndexInput. */ - public static boolean isFSIndexInput(IndexInput is) { + public static boolean isFSIndexInput(BareInput is) { return is instanceof FSIndexInput; } @@ -38,7 +38,7 @@ */ public static boolean isFSIndexInputClone(IndexInput is) { if (isFSIndexInput(is)) { - return ((FSIndexInput) is).isClone; + return ((FSDirectory.FSBareInput)((FSIndexInput) is).input).isClone; } else { return false; } @@ -51,12 +51,14 @@ * FSIndexInput that owns this descriptor is closed. However, the * descriptor may possibly become invalid in other ways as well. */ - public static boolean isFSIndexInputOpen(IndexInput is) + public static boolean isFSIndexInputOpen(BareInput is) throws IOException { if (isFSIndexInput(is)) { FSIndexInput fis = (FSIndexInput) is; return fis.isFDValid(); + } else if (is instanceof FSDirectory.FSBareInput) { + return ((FSDirectory.FSBareInput)is).isFDValid(); } else { return false; } Index: test/org/apache/lucene/index/TestCompoundFile.java =================================================================== --- test/org/apache/lucene/index/TestCompoundFile.java (revision 588053) +++ test/org/apache/lucene/index/TestCompoundFile.java (working copy) @@ -23,6 +23,9 @@ import org.apache.lucene.util.LuceneTestCase; import junit.framework.TestSuite; import junit.textui.TestRunner; + +import org.apache.lucene.store.BareInput; +import org.apache.lucene.store.BufferPooledIndexInput; import org.apache.lucene.store.IndexOutput; import org.apache.lucene.store.Directory; import org.apache.lucene.store.IndexInput; @@ -358,7 +361,11 @@ CompoundFileReader.CSIndexInput cis = (CompoundFileReader.CSIndexInput) is; - return _TestHelper.isFSIndexInputOpen(cis.base); + if (cis.base instanceof BufferPooledIndexInput) { + return _TestHelper.isFSIndexInputOpen(((BufferPooledIndexInput)cis.base).getInput()); + } else { + return _TestHelper.isFSIndexInputOpen(cis.base); + } } else { return false; } @@ -374,7 +381,6 @@ // this test only works for FSIndexInput if (_TestHelper.isFSIndexInput(expected)) { - assertTrue(_TestHelper.isFSIndexInputOpen(expected)); IndexInput one = cr.openInput("f11"); @@ -415,6 +421,50 @@ expected.seek(0); two.seek(0); //assertSameStreams("basic clone two/4", expected, two); + } else if (expected instanceof BufferPooledIndexInput) { + BareInput expectedInput = ((BufferPooledIndexInput)expected).getInput(); + assertTrue(_TestHelper.isFSIndexInputOpen(expectedInput)); + + IndexInput one = cr.openInput("f11"); + assertTrue(isCSIndexInputOpen(one)); + + IndexInput two = (IndexInput) one.clone(); + assertTrue(isCSIndexInputOpen(two)); + + assertSameStreams("basic clone one", expected, one); + expected.seek(0); + assertSameStreams("basic clone two", expected, two); + + // Now close the first stream + one.close(); + assertTrue("Only close when cr is closed", isCSIndexInputOpen(one)); + + // The following should really fail since we couldn't expect to + // access a file once close has been called on it (regardless of + // buffering and/or clone magic) + expected.seek(0); + two.seek(0); + assertSameStreams("basic clone two/2", expected, two); + + + // Now close the compound reader + cr.close(); + assertFalse("Now closed one", isCSIndexInputOpen(one)); + assertFalse("Now closed two", isCSIndexInputOpen(two)); + + // The following may also fail since the compound stream is closed + expected.seek(0); + two.seek(0); + //assertSameStreams("basic clone two/3", expected, two); + + + // Now close the second clone + two.close(); + expected.seek(0); + two.seek(0); + //assertSameStreams("basic clone two/4", expected, two); + } else { + cr.close(); } expected.close(); Index: java/org/apache/lucene/index/IndexFileNames.java =================================================================== --- java/org/apache/lucene/index/IndexFileNames.java (revision 588053) +++ java/org/apache/lucene/index/IndexFileNames.java (working copy) @@ -23,7 +23,7 @@ * @author Bernhard Messer * @version $rcs = ' $Id: Exp $ ' ; */ -final class IndexFileNames { +public final class IndexFileNames { /** Name of the index segment file */ static final String SEGMENTS = "segments"; @@ -163,6 +163,10 @@ VECTORS_FIELDS_EXTENSION }; + static final String WRITABLE_FILE_NAMES[] = new String[] { + SEGMENTS_GEN + }; + /** * Computes the full file name from base, extension and * generation. If the generation is -1, the file name is @@ -196,4 +200,12 @@ return true; return false; } + + public static final boolean isWritableFile(String fileName) { + for (int i=0; i>> 32)); // hash code of long + } + + public boolean equals(Object other) { + BufferKey otherKey = (BufferKey) other; + return position == otherKey.position && source.equals(otherKey.source); + } + + public int hashCode() { + return hashCode; + } + } + + static class Buffer { + BufferKey key; + final byte[] buffer; + + Buffer prev; + Buffer next; + + Buffer(int bufferSize) { + key = new BufferKey(null, -1); + buffer = new byte[bufferSize]; + } + } + + final private int bufferSize; + + final private Buffer[] pool; + final private HashMap map; // map from (file, positione) to buffer + + private Buffer head; // remove from head - least recently used + private Buffer tail; // add to tail - most recently used + + private BufferKey checkKey = new BufferKey(null, -1); + + private Stats stats = new Stats(); + + public BufferPoolLRU(int bufferSize, int numBuffers) { + this.bufferSize = bufferSize; + + pool = new Buffer[numBuffers]; + map = new HashMap(); + + for (int i = 0; i < numBuffers; i++) { + pool[i] = new Buffer(bufferSize); + if (i != 0) { + pool[i - 1].next = pool[i]; + pool[i].prev = pool[i - 1]; + } + } + + head = pool[0]; + tail = pool[numBuffers - 1]; + } + + public int getBufferSize() { + return bufferSize; + } + + public Stats getStats() { + return stats; + } + + public void close() throws IOException { + // only do verification + // set pool and map to null? + int count = head != null ? 1 : 0; + while (head != tail) { + head = head.next; + count++; + } + if (count != pool.length) { + throw new IOException("Buffer pool internal error: expected " + + pool.length + " buffers, actual " + count + " buffers"); + } + } + + // assume position + len <= input.length, i.e. not read beyond EOF + public void read(String source, long position, byte[] b, int offset, int len, + BareInput input) throws IOException { + + if (len > bufferSize) { + // read directly into b + readInput(input, position, b, offset, len); + return; + } + + long quotient = position / bufferSize; + long bufferPos = quotient * bufferSize; + int remainder = (int) (position - bufferPos); + + int oneLen = len; // length for this read, may need two reads + boolean more = false; + if (len > bufferSize - remainder) { + oneLen = bufferSize - remainder; + more = true; + } + + Buffer buffer = null; + boolean success = false; + + synchronized (this) { + stats.numReads++; + checkKey.set(source, bufferPos); + buffer = (Buffer)map.get(checkKey); + + if (buffer != null) { + System.arraycopy(buffer.buffer, remainder, b, offset, oneLen); + success = true; + stats.numHits++; + + remove(buffer); + addTail(buffer); + } + } + + if (!success) { + synchronized (this) { + buffer = removeHead(); + if (buffer != null) { + map.remove(buffer.key); + } + } + + if (buffer != null) { + try { + // read into buffer + readInput(input, bufferPos, buffer.buffer, 0, + (bufferPos + bufferSize <= input.length()) ? bufferSize + : (int) (input.length() - bufferPos)); + buffer.key.set(source, bufferPos); + } catch (IOException e) { + addHead(buffer); + throw e; + } + + System.arraycopy(buffer.buffer, remainder, b, offset, oneLen); + + synchronized (this) { + map.put(buffer.key, buffer); + addTail(buffer); + } + } else { + // no more buffer, read directly into b + readInput(input, position, b, offset, len); + more = false; + } + } + + if (more) { + read(source, position + oneLen, b, offset + oneLen, len - oneLen, input); + } + } + + // remove buffer from doubly linked list + private void remove(Buffer buffer) { + // head != null, tail != null + if (buffer.prev != null) { // buffer != head + buffer.prev.next = buffer.next; + } else { + head = buffer.next; + } + if (buffer.next != null) { // buffer != tail + buffer.next.prev = buffer.prev; + } else { + tail = buffer.prev; + } + buffer.prev = null; + buffer.next = null; + } + + // remove from head - LRU + private Buffer removeHead() { + Buffer buffer = head; + if (head != null) { + remove(head); + } + return buffer; + } + + // add to tail - MRU + private void addTail(Buffer buffer) { + if (head == null) { // tail == null + head = tail = buffer; + // buffer.prev = buffer.next = null; + } else { + buffer.prev = tail; + // buffer.next = null; + tail.next = buffer; + tail = buffer; + } + } + + // add to head - LRU + private void addHead(Buffer buffer) { + if (head == null) { // tail == null + head = tail = buffer; + // buffer.prev = buffer.next = null; + } else { + // buffer.prev = null; + buffer.next = head; + head.prev = buffer; + head = buffer; + } + } + + protected void readInput(BareInput input, long position, byte[] b, + int offset, int len) throws IOException { + if (position != input.getFilePointer()) { + input.seek(position); + } + input.readBytes(b, offset, len); + } +} Index: java/org/apache/lucene/store/IndexInput.java =================================================================== --- java/org/apache/lucene/store/IndexInput.java (revision 588053) +++ java/org/apache/lucene/store/IndexInput.java (working copy) @@ -23,7 +23,7 @@ * random-access input stream. Used for all Lucene index input operations. * @see Directory */ -public abstract class IndexInput implements Cloneable { +public abstract class IndexInput implements BareInput, Cloneable { private char[] chars; // used by readString() /** Reads and returns a single byte. Index: java/org/apache/lucene/store/BareInput.java =================================================================== --- java/org/apache/lucene/store/BareInput.java (revision 0) +++ java/org/apache/lucene/store/BareInput.java (revision 0) @@ -0,0 +1,42 @@ +package org.apache.lucene.store; + +import java.io.IOException; + +public interface BareInput extends Cloneable { + + /** Reads a specified number of bytes into an array at the specified offset. + * @param b the array to read bytes into + * @param offset the offset in the array to start storing bytes + * @param len the number of bytes to read + * @see IndexOutput#writeBytes(byte[],int) + */ + void readBytes(byte[] b, int offset, int len) throws IOException; + + /** Closes the stream to futher operations. */ + void close() throws IOException; + + /** Returns the current position in this file, where the next read will + * occur. + * @see #seek(long) + */ + long getFilePointer(); + + /** Sets current position in this file, where the next read will occur. + * @see #getFilePointer() + */ + void seek(long pos) throws IOException; + + /** The number of bytes in the file. */ + long length(); + + /** Returns a clone of this stream. + * + *

Clones of a stream access the same data, and are positioned at the same + * point as the stream they were cloned from. + * + *

Expert: Subclasses must ensure that clones may be positioned at + * different points in the input from each other and from the stream they + * were cloned from. + */ + Object clone(); +} Index: java/org/apache/lucene/store/BufferedIndexInput.java =================================================================== --- java/org/apache/lucene/store/BufferedIndexInput.java (revision 588053) +++ java/org/apache/lucene/store/BufferedIndexInput.java (working copy) @@ -20,7 +20,7 @@ import java.io.IOException; /** Base implementation class for buffered {@link IndexInput}. */ -public abstract class BufferedIndexInput extends IndexInput { +public abstract class BufferedIndexInput extends IndexInput implements BareInput { /** Default buffer size */ public static final int BUFFER_SIZE = 1024; Index: java/org/apache/lucene/store/FSDirectory.java =================================================================== --- java/org/apache/lucene/store/FSDirectory.java (revision 588053) +++ java/org/apache/lucene/store/FSDirectory.java (working copy) @@ -26,6 +26,7 @@ import java.security.NoSuchAlgorithmException; import java.util.Hashtable; +import org.apache.lucene.index.IndexFileNames; import org.apache.lucene.index.IndexFileNameFilter; // Used only for WRITE_LOCK_NAME in deprecated create=true case: @@ -136,8 +137,13 @@ * @return the FSDirectory for the named file. */ public static FSDirectory getDirectory(String path) throws IOException { - return getDirectory(new File(path), null); + return getDirectory(new File(path), null, 0, 0); } + // a buffer pool won't be created if the dir object already exists + public static FSDirectory getDirectory(String path, int bufferSize, + int numBuffers) throws IOException { + return getDirectory(new File(path), null, bufferSize, numBuffers); + } /** Returns the directory instance for the named location. * @param path the path to the directory. @@ -146,24 +152,34 @@ * @return the FSDirectory for the named file. */ public static FSDirectory getDirectory(String path, LockFactory lockFactory) throws IOException { - return getDirectory(new File(path), lockFactory); + return getDirectory(new File(path), lockFactory, 0, 0); } + // a buffer pool won't be created if the dir object already exists + public static FSDirectory getDirectory(String path, LockFactory lockFactory, + int bufferSize, int numBuffers) throws IOException { + return getDirectory(new File(path), lockFactory, bufferSize, numBuffers); + } /** Returns the directory instance for the named location. * @param file the path to the directory. * @return the FSDirectory for the named file. */ public static FSDirectory getDirectory(File file) throws IOException { - return getDirectory(file, null); + return getDirectory(file, null, 0, 0); } + // a buffer pool won't be created if the dir object already exists + public static FSDirectory getDirectory(File file, int bufferSize, + int numBuffers) throws IOException { + return getDirectory(file, null, bufferSize, numBuffers); + } /** Returns the directory instance for the named location. * @param file the path to the directory. * @param lockFactory instance of {@link LockFactory} providing the * locking implementation. * @return the FSDirectory for the named file. */ - public static FSDirectory getDirectory(File file, LockFactory lockFactory) - throws IOException + public static FSDirectory getDirectory(File file, LockFactory lockFactory, + int bufferSize, int numBuffers) throws IOException { file = new File(file.getCanonicalPath()); @@ -183,7 +199,7 @@ } catch (Exception e) { throw new RuntimeException("cannot load FSDirectory class: " + e.toString(), e); } - dir.init(file, lockFactory); + dir.init(file, lockFactory, bufferSize, numBuffers); DIRECTORIES.put(file, dir); } else { // Catch the case where a Directory is pulled from the cache, but has a @@ -224,7 +240,7 @@ public static FSDirectory getDirectory(File file, boolean create) throws IOException { - FSDirectory dir = getDirectory(file, null); + FSDirectory dir = getDirectory(file, null, 0, 0); // This is now deprecated (creation should only be done // by IndexWriter): @@ -250,11 +266,13 @@ } private File directory = null; + private BufferPool pool = null; private int refCount; protected FSDirectory() {}; // permit subclassing - private void init(File path, LockFactory lockFactory) throws IOException { + private void init(File path, LockFactory lockFactory, int bufferSize, + int numBuffers) throws IOException { // Set up lockFactory with cascaded defaults: if an instance was passed in, // use that; else if locks are disabled, use NoLockFactory; else if the @@ -263,6 +281,10 @@ directory = path; + if (bufferSize >= BufferedIndexInput.BUFFER_SIZE && numBuffers > 1) { + pool = new BufferPoolLRU(bufferSize, numBuffers); + } + boolean doClearLockID = false; if (lockFactory == null) { @@ -315,6 +337,13 @@ } } + public BufferPool.Stats getBufferPoolStats() { + if (pool != null) { + return pool.getStats(); + } + return null; + } + /** Returns an array of strings, one for each Lucene index file in the directory. */ public String[] list() { return directory.list(IndexFileNameFilter.getFilter()); @@ -437,12 +466,23 @@ // Inherit javadoc public IndexInput openInput(String name) throws IOException { - return new FSIndexInput(new File(directory, name)); + FSBareInput input = new FSBareInput(new File(directory, name)); + if (pool == null || IndexFileNames.isWritableFile(name)) { + return new FSIndexInput(input); + } else { + return new BufferPooledIndexInput(name, input, pool); + } } // Inherit javadoc public IndexInput openInput(String name, int bufferSize) throws IOException { - return new FSIndexInput(new File(directory, name), bufferSize); + FSBareInput input = new FSBareInput(new File(directory, name)); + if (pool == null || IndexFileNames.isWritableFile(name)) { + return new FSIndexInput(input, bufferSize); + } else { + // TODO should warn if bufferSize > pool.getBufferSize() + return new BufferPooledIndexInput(name, input, pool); + } } /** @@ -476,11 +516,14 @@ } /** Closes the store to future operations. */ - public synchronized void close() { + public synchronized void close() throws IOException { if (--refCount <= 0) { synchronized (DIRECTORIES) { DIRECTORIES.remove(directory); } + if (pool != null) { + pool.close(); + } } } @@ -493,28 +536,28 @@ return this.getClass().getName() + "@" + directory; } - protected static class FSIndexInput extends BufferedIndexInput { - + protected static class FSBareInput implements BareInput { + private static class Descriptor extends RandomAccessFile { // remember if the file is open, so that we don't try to close it // more than once private boolean isOpen; long position; final long length; - + public Descriptor(File file, String mode) throws IOException { super(file, mode); isOpen=true; length=length(); } - + public void close() throws IOException { if (isOpen) { isOpen=false; super.close(); } } - + protected void finalize() throws Throwable { try { close(); @@ -523,57 +566,65 @@ } } } - + private final Descriptor file; + long position; boolean isClone; - - public FSIndexInput(File path) throws IOException { - this(path, BufferedIndexInput.BUFFER_SIZE); - } - - public FSIndexInput(File path, int bufferSize) throws IOException { - super(bufferSize); + + public FSBareInput(File path) throws IOException { file = new Descriptor(path, "r"); } - - /** IndexInput methods */ - protected void readInternal(byte[] b, int offset, int len) + + /** BareInput methods */ + public void readBytes(byte[] b, int offset, int len) throws IOException { synchronized (file) { - long position = getFilePointer(); - if (position != file.position) { - file.seek(position); - file.position = position; + try { + if (position != file.position) { + file.seek(position); + file.position = position; + } + int total = 0; + do { + int i = file.read(b, offset + total, len - total); + if (i == -1) + throw new IOException("read past EOF"); + file.position += i; + total += i; + } while (total < len); + } finally { + position = file.position; } - int total = 0; - do { - int i = file.read(b, offset+total, len-total); - if (i == -1) - throw new IOException("read past EOF"); - file.position += i; - total += i; - } while (total < len); } } - + public void close() throws IOException { // only close the file if this is not a clone if (!isClone) file.close(); } - - protected void seekInternal(long position) { + + public long getFilePointer() { + return position; } - + + public void seek(long position) { + this.position = position; + } + public long length() { return file.length; } - + public Object clone() { - FSIndexInput clone = (FSIndexInput)super.clone(); + FSBareInput clone = null; + try { + clone = (FSBareInput)super.clone(); + } catch (CloneNotSupportedException e) { + } clone.isClone = true; return clone; } - + /** Method used for testing. Returns true if the underlying * file descriptor is valid. */ @@ -582,6 +633,50 @@ } } + protected static class FSIndexInput extends BufferedIndexInput { + BareInput input; + + public FSIndexInput(BareInput input) throws IOException { + this(input, BufferedIndexInput.BUFFER_SIZE); + } + + public FSIndexInput(BareInput input, int bufferSize) throws IOException { + super(bufferSize); + this.input = input; + } + + /** IndexInput methods */ + protected void readInternal(byte[] b, int offset, int len) + throws IOException { + input.seek(getFilePointer()); + input.readBytes(b, offset, len); + } + + public void close() throws IOException { + input.close(); + } + + protected void seekInternal(long position) throws IOException { + } + + public long length() { + return input.length(); + } + + public Object clone() { + FSIndexInput clone = (FSIndexInput)super.clone(); + clone.input = (BareInput)input.clone(); + return clone; + } + + /** Method used for testing. Returns true if the underlying + * file descriptor is valid. + */ + boolean isFDValid() throws IOException { + return ((FSBareInput)input).isFDValid(); + } + } + protected static class FSIndexOutput extends BufferedIndexOutput { RandomAccessFile file = null;