Add a method to ObjectInserter to read back inserted objects

In the DFS implementation, flushing an inserter writes a new pack to
the storage system and is potentially very slow, but was the only way
to ensure previously-inserted objects were available.  For some tasks,
like performing a series of three-way merges, the total size of all
inserted objects may be small enough to avoid flushing the in-memory
buffered data.

DfsOutputStream already provides a read method to read back from the
not-yet-flushed data, so use this to provide an ObjectReader in the
DFS case.

In the file-backed case, objects are written out loosely on the fly,
so the implementation can just return the existing WindowCursor.

Change-Id: I454fdfb88f4d215e31b7da2b2a069853b197b3dd

6 files changed, 330 insertions, 2 deletions

diff --git a/org.eclipse.jgit/.settings/.api_filters b/org.eclipse.jgit/.settings/.api_filters
index 01c243417a..dc0133c365 100644
--- a/org.eclipse.jgit/.settings/.api_filters
+++ b/org.eclipse.jgit/.settings/.api_filters
@@ -8,6 +8,14 @@
             </message_arguments>
         </filter>
     </resource>
+    <resource path="src/org/eclipse/jgit/lib/ObjectInserter.java" type="org.eclipse.jgit.lib.ObjectInserter">
+        <filter id="336695337">
+            <message_arguments>
+                <message_argument value="org.eclipse.jgit.lib.ObjectInserter"/>
+                <message_argument value="newReader()"/>
+            </message_arguments>
+        </filter>
+    </resource>
     <resource path="src/org/eclipse/jgit/merge/ResolveMerger.java" type="org.eclipse.jgit.merge.ResolveMerger">
         <filter comment="Doesn't break consumers. Breaking providers is allowed also in minor versions." id="338792546">
             <message_arguments>
diff --git a/org.eclipse.jgit/resources/org/eclipse/jgit/internal/storage/dfs/DfsText.properties b/org.eclipse.jgit/resources/org/eclipse/jgit/internal/storage/dfs/DfsText.properties
index 2c4bd06a33..4bbc4cc71a 100644
--- a/org.eclipse.jgit/resources/org/eclipse/jgit/internal/storage/dfs/DfsText.properties
+++ b/org.eclipse.jgit/resources/org/eclipse/jgit/internal/storage/dfs/DfsText.properties
@@ -1,4 +1,6 @@
 cannotReadIndex=Cannot read index {0}
+cannotReadBackDelta=Cannot read delta type {0}
 shortReadOfBlock=Short read of block at {0} in pack {1}; expected {2} bytes, received only {3}
 shortReadOfIndex=Short read of index {0}
+unexpectedEofInPack=Unexpected EOF in partially created pack
 willNotStoreEmptyPack=Cannot store empty pack
diff --git a/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsInserter.java b/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsInserter.java
index 6be6509b46..22a03dd8e6 100644
--- a/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsInserter.java
+++ b/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsInserter.java
@@ -45,24 +45,44 @@ package org.eclipse.jgit.internal.storage.dfs;
 
 import static org.eclipse.jgit.internal.storage.pack.PackExt.INDEX;
 import static org.eclipse.jgit.internal.storage.pack.PackExt.PACK;
+import static org.eclipse.jgit.lib.Constants.OBJ_OFS_DELTA;
+import static org.eclipse.jgit.lib.Constants.OBJ_REF_DELTA;
 
+import java.io.BufferedInputStream;
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.nio.ByteBuffer;
 import java.security.MessageDigest;
+import java.text.MessageFormat;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.zip.CRC32;
+import java.util.zip.DataFormatException;
 import java.util.zip.Deflater;
 import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
 
+import org.eclipse.jgit.errors.CorruptObjectException;
+import org.eclipse.jgit.errors.LargeObjectException;
+import org.eclipse.jgit.internal.JGitText;
 import org.eclipse.jgit.internal.storage.file.PackIndex;
 import org.eclipse.jgit.internal.storage.file.PackIndexWriter;
+import org.eclipse.jgit.internal.storage.pack.PackExt;
+import org.eclipse.jgit.lib.AbbreviatedObjectId;
+import org.eclipse.jgit.lib.AnyObjectId;
 import org.eclipse.jgit.lib.Constants;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.lib.ObjectIdOwnerMap;
 import org.eclipse.jgit.lib.ObjectInserter;
+import org.eclipse.jgit.lib.ObjectLoader;
+import org.eclipse.jgit.lib.ObjectReader;
+import org.eclipse.jgit.lib.ObjectStream;
 import org.eclipse.jgit.transport.PackedObjectInfo;
 import org.eclipse.jgit.util.BlockList;
 import org.eclipse.jgit.util.IO;
@@ -76,6 +96,7 @@ public class DfsInserter extends ObjectInserter {
 	private static final int INDEX_VERSION = 2;
 
 	private final DfsObjDatabase db;
+	private int compression = Deflater.BEST_COMPRESSION;
 
 	private List<PackedObjectInfo> objectList;
 	private ObjectIdOwnerMap<PackedObjectInfo> objectMap;
@@ -96,12 +117,21 @@ public class DfsInserter extends ObjectInserter {
 		this.db = db;
 	}
 
+	void setCompressionLevel(int compression) {
+		this.compression = compression;
+	}
+
 	@Override
 	public DfsPackParser newPackParser(InputStream in) throws IOException {
 		return new DfsPackParser(db, this, in);
 	}
 
 	@Override
+	public ObjectReader newReader() {
+		return new Reader();
+	}
+
+	@Override
 	public ObjectId insert(int type, byte[] data, int off, int len)
 			throws IOException {
 		ObjectId id = idFor(type, data, off, len);
@@ -309,7 +339,7 @@ public class DfsInserter extends ObjectInserter {
 			hdrBuf = new byte[32];
 			md = Constants.newMessageDigest();
 			crc32 = new CRC32();
-			deflater = new Deflater(Deflater.BEST_COMPRESSION);
+			deflater = new Deflater(compression);
 			compress = new DeflaterOutputStream(this, deflater, 8192);
 
 			int size = out.blockSize();
@@ -403,10 +433,266 @@ public class DfsInserter extends ObjectInserter {
 			return packHash;
 		}
 
+		int read(long pos, byte[] dst, int ptr, int cnt) throws IOException {
+			int r = 0;
+			while (pos < currPos && r < cnt) {
+				DfsBlock b = getOrLoadBlock(pos);
+				int n = b.copy(pos, dst, ptr + r, cnt - r);
+				pos += n;
+				r += n;
+			}
+			if (currPos <= pos && r < cnt) {
+				int s = (int) (pos - currPos);
+				int n = Math.min(currPtr - s, cnt - r);
+				System.arraycopy(currBuf, s, dst, ptr + r, n);
+				r += n;
+			}
+			return r;
+		}
+
+		byte[] inflate(DfsReader ctx, long pos, int len) throws IOException,
+				DataFormatException {
+			byte[] dstbuf;
+			try {
+				dstbuf = new byte[len];
+			} catch (OutOfMemoryError noMemory) {
+				return null; // Caller will switch to large object streaming.
+			}
+
+			Inflater inf = ctx.inflater();
+			DfsBlock b = setInput(inf, pos);
+			for (int dstoff = 0;;) {
+				int n = inf.inflate(dstbuf, dstoff, dstbuf.length - dstoff);
+				if (n > 0)
+					dstoff += n;
+				else if (inf.needsInput() && b != null) {
+					pos += b.remaining(pos);
+					b = setInput(inf, pos);
+				} else if (inf.needsInput())
+					throw new EOFException(DfsText.get().unexpectedEofInPack);
+				else if (inf.finished())
+					return dstbuf;
+				else
+					throw new DataFormatException();
+			}
+		}
+
+		private DfsBlock setInput(Inflater inf, long pos) throws IOException {
+			if (pos < currPos) {
+				DfsBlock b = getOrLoadBlock(pos);
+				b.setInput(inf, pos);
+				return b;
+			}
+			inf.setInput(currBuf, (int) (pos - currPos), currPtr);
+			return null;
+		}
+
+		private DfsBlock getOrLoadBlock(long pos) throws IOException {
+			long s = toBlockStart(pos);
+			DfsBlock b = cache.get(packKey, s);
+			if (b != null)
+				return b;
+
+			byte[] d = new byte[blockSize];
+			for (int p = 0; p < blockSize;) {
+				int n = out.read(s + p, ByteBuffer.wrap(d, p, blockSize - p));
+				if (n <= 0)
+					throw new EOFException(DfsText.get().unexpectedEofInPack);
+				p += n;
+			}
+			b = new DfsBlock(packKey, s, d);
+			cache.put(b);
+			return b;
+		}
+
+		private long toBlockStart(long pos) {
+			return (pos / blockSize) * blockSize;
+		}
+
 		@Override
 		public void close() throws IOException {
 			deflater.end();
 			out.close();
 		}
 	}
+
+	private class Reader extends ObjectReader {
+		private final DfsReader ctx = new DfsReader(db);
+
+		@Override
+		public ObjectReader newReader() {
+			return db.newReader();
+		}
+
+		@Override
+		public Collection<ObjectId> resolve(AbbreviatedObjectId id)
+				throws IOException {
+			Collection<ObjectId> stored = ctx.resolve(id);
+			if (objectList == null)
+				return stored;
+
+			Set<ObjectId> r = new HashSet<ObjectId>(stored.size() + 2);
+			r.addAll(stored);
+			for (PackedObjectInfo obj : objectList) {
+				if (id.prefixCompare(obj) == 0)
+					r.add(obj.copy());
+			}
+			return r;
+		}
+
+		@Override
+		public ObjectLoader open(AnyObjectId objectId, int typeHint)
+				throws IOException {
+			if (objectMap == null)
+				return ctx.open(objectId, typeHint);
+
+			PackedObjectInfo obj = objectMap.get(objectId);
+			if (obj == null)
+				return ctx.open(objectId, typeHint);
+
+			byte[] buf = buffer();
+			int cnt = packOut.read(obj.getOffset(), buf, 0, 20);
+			if (cnt <= 0)
+					throw new EOFException(DfsText.get().unexpectedEofInPack);
+
+			int c = buf[0] & 0xff;
+			int type = (c >> 4) & 7;
+			if (type == OBJ_OFS_DELTA || type == OBJ_REF_DELTA)
+				throw new IOException(MessageFormat.format(
+						DfsText.get().cannotReadBackDelta, Integer.toString(type)));
+
+			long sz = c & 0x0f;
+			int ptr = 1;
+			int shift = 4;
+			while ((c & 0x80) != 0) {
+				if (ptr >= cnt)
+					throw new EOFException(DfsText.get().unexpectedEofInPack);
+				c = buf[ptr++] & 0xff;
+				sz += ((long) (c & 0x7f)) << shift;
+				shift += 7;
+			}
+
+			long zpos = obj.getOffset() + ptr;
+			if (sz < ctx.getStreamFileThreshold()) {
+				byte[] data = inflate(obj, zpos, (int) sz);
+				if (data != null)
+					return new ObjectLoader.SmallObject(type, data);
+			}
+			return new StreamLoader(obj.copy(), type, sz, packKey, zpos);
+		}
+
+		private byte[] inflate(PackedObjectInfo obj, long zpos, int sz)
+				throws IOException, CorruptObjectException {
+			try {
+				return packOut.inflate(ctx, zpos, sz);
+			} catch (DataFormatException dfe) {
+				CorruptObjectException coe = new CorruptObjectException(
+						MessageFormat.format(
+								JGitText.get().objectAtHasBadZlibStream,
+								Long.valueOf(obj.getOffset()),
+								packDsc.getFileName(PackExt.PACK)));
+				coe.initCause(dfe);
+				throw coe;
+			}
+		}
+
+		@Override
+		public Set<ObjectId> getShallowCommits() throws IOException {
+			return ctx.getShallowCommits();
+		}
+
+		@Override
+		public void release() {
+			ctx.release();
+		}
+	}
+
+	private class StreamLoader extends ObjectLoader {
+		private final ObjectId id;
+		private final int type;
+		private final long size;
+
+		private final DfsPackKey srcPack;
+		private final long pos;
+
+		StreamLoader(ObjectId id, int type, long sz,
+				DfsPackKey key, long pos) {
+			this.id = id;
+			this.type = type;
+			this.size = sz;
+			this.srcPack = key;
+			this.pos = pos;
+		}
+
+		@Override
+		public ObjectStream openStream() throws IOException {
+			final DfsReader ctx = new DfsReader(db);
+			if (srcPack != packKey) {
+				try {
+					// Post DfsInserter.flush() use the normal code path.
+					// The newly created pack is registered in the cache.
+					return ctx.open(id, type).openStream();
+				} finally {
+					ctx.release();
+				}
+			}
+
+			int bufsz = 8192;
+			final Inflater inf = ctx.inflater();
+			return new ObjectStream.Filter(type,
+					size, new BufferedInputStream(new InflaterInputStream(
+							new ReadBackStream(pos), inf, bufsz), bufsz)) {
+				@Override
+				public void close() throws IOException {
+					ctx.release();
+					super.close();
+				}
+			};
+		}
+
+		@Override
+		public int getType() {
+			return type;
+		}
+
+		@Override
+		public long getSize() {
+			return size;
+		}
+
+		@Override
+		public boolean isLarge() {
+			return true;
+		}
+
+		@Override
+		public byte[] getCachedBytes() throws LargeObjectException {
+			throw new LargeObjectException.ExceedsLimit(
+					db.getReaderOptions().getStreamFileThreshold(), size);
+		}
+	}
+
+	private final class ReadBackStream extends InputStream {
+		private long pos;
+
+		ReadBackStream(long offset) {
+			pos = offset;
+		}
+
+		@Override
+		public int read() throws IOException {
+			byte[] b = new byte[1];
+			int n = read(b);
+			return n == 1 ? b[0] & 0xff : -1;
+		}
+
+		@Override
+		public int read(byte[] buf, int ptr, int len) throws IOException {
+			int n = packOut.read(pos, buf, ptr, len);
+			if (n > 0) {
+				pos += n;
+			}
+			return n;
+		}
+	}
 }
diff --git a/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsText.java b/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsText.java
index dedcab04d8..862454759e 100644
--- a/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsText.java
+++ b/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/dfs/DfsText.java
@@ -55,7 +55,9 @@ public class DfsText extends TranslationBundle {
 
 	// @formatter:off
 	/***/ public String cannotReadIndex;
+	/***/ public String cannotReadBackDelta;
 	/***/ public String shortReadOfBlock;
 	/***/ public String shortReadOfIndex;
+	/***/ public String unexpectedEofInPack;
 	/***/ public String willNotStoreEmptyPack;
 }
diff --git a/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/file/ObjectDirectoryInserter.java b/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/file/ObjectDirectoryInserter.java
index c3e6ad9630..812c899a8f 100644
--- a/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/file/ObjectDirectoryInserter.java
+++ b/org.eclipse.jgit/src/org/eclipse/jgit/internal/storage/file/ObjectDirectoryInserter.java
@@ -63,6 +63,7 @@ import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.lib.Constants;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.lib.ObjectInserter;
+import org.eclipse.jgit.lib.ObjectReader;
 import org.eclipse.jgit.transport.PackParser;
 import org.eclipse.jgit.util.FileUtils;
 import org.eclipse.jgit.util.IO;
@@ -131,8 +132,13 @@ class ObjectDirectoryInserter extends ObjectInserter {
 	}
 
 	@Override
+	public ObjectReader newReader() {
+		return new WindowCursor(db);
+	}
+
+	@Override
 	public void flush() throws IOException {
-		// Do nothing. Objects are immediately visible.
+		// Do nothing. Loose objects are immediately visible.
 	}
 
 	@Override
diff --git a/org.eclipse.jgit/src/org/eclipse/jgit/lib/ObjectInserter.java b/org.eclipse.jgit/src/org/eclipse/jgit/lib/ObjectInserter.java
index 0bed0dd9ed..5c13ef3548 100644
--- a/org.eclipse.jgit/src/org/eclipse/jgit/lib/ObjectInserter.java
+++ b/org.eclipse.jgit/src/org/eclipse/jgit/lib/ObjectInserter.java
@@ -81,6 +81,11 @@ public abstract class ObjectInserter {
 		}
 
 		@Override
+		public ObjectReader newReader() {
+			throw new UnsupportedOperationException();
+		}
+
+		@Override
 		public void flush() throws IOException {
 			// Do nothing.
 		}
@@ -136,6 +141,10 @@ public abstract class ObjectInserter {
 			return delegate().newPackParser(in);
 		}
 
+		public ObjectReader newReader() {
+			return delegate().newReader();
+		}
+
 		public void flush() throws IOException {
 			delegate().flush();
 		}
@@ -381,6 +390,21 @@ public abstract class ObjectInserter {
 	public abstract PackParser newPackParser(InputStream in) throws IOException;
 
 	/**
+	 * Open a reader for objects that may have been written by this inserter.
+	 * <p>
+	 * The returned reader allows the calling thread to read back recently
+	 * inserted objects without first calling {@code flush()} to make them
+	 * visible to the repository. The returned reader should only be used from
+	 * the same thread as the inserter. Objects written by this inserter may not
+	 * be visible to {@code this.newReader().newReader()}.
+	 *
+	 * @since 3.5
+	 * @return reader for any object, including an object recently inserted by
+	 *         this inserter since the last flush.
+	 */
+	public abstract ObjectReader newReader();
+
+	/**
 	 * Make all inserted objects visible.
 	 * <p>
 	 * The flush may take some period of time to make the objects available to