Use core.streamFileThreshold to set our streaming limit

We default this to 1 MiB for now, but we allow users to modify
it through the Repository's configuration file to be a different
value.  A new repository listener is used to identify when the
setting has been updated and trigger a reconfiguration of any
active ObjectReaders.

To prevent a horrible explosion we cap core.streamFileThreshold
at no more than 1/4 of the maximum JVM heap size.  We do this
because we need at least 2 byte arrays equal in size to the
stream threshold for the worst case delta inflation scenario,
and our host application probably also needs some amount of the
heap for their working set size.

Change-Id: I103b3a541dc970bbf1a6d92917a12c5a1ee34d6c
Signed-off-by: Shawn O. Pearce <spearce@spearce.org>

org.eclipse.jgit.test/tst/org/eclipse/jgit/storage/file/PackFileTest.java

@@ -118,7 +118,7 @@ public class PackFileTest extends LocalDiskRepositoryTestCase {
 
 	public void testWhole_LargeObject() throws Exception {
 		final int type = Constants.OBJ_BLOB;
-		byte[] data = rng.nextBytes(UnpackedObject.LARGE_OBJECT + 5);
+		byte[] data = rng.nextBytes(ObjectLoader.STREAM_THRESHOLD + 5);
 		RevBlob id = tr.blob(data);
 		tr.branch("master").commit().add("A", id).create();
 		tr.packAndPrune();
@@ -209,7 +209,7 @@ public class PackFileTest extends LocalDiskRepositoryTestCase {
 
 	public void testDelta_LargeObjectChain() throws Exception {
 		ObjectInserter.Formatter fmt = new ObjectInserter.Formatter();
-		byte[] data0 = new byte[UnpackedObject.LARGE_OBJECT + 5];
+		byte[] data0 = new byte[ObjectLoader.STREAM_THRESHOLD + 5];
 		Arrays.fill(data0, (byte) 0xf3);
 		ObjectId id0 = fmt.idFor(Constants.OBJ_BLOB, data0);
 
@@ -277,12 +277,12 @@ public class PackFileTest extends LocalDiskRepositoryTestCase {
 		Arrays.fill(data0, (byte) 0xf3);
 		ObjectId id0 = fmt.idFor(Constants.OBJ_BLOB, data0);
 
-		byte[] data3 = rng.nextBytes(UnpackedObject.LARGE_OBJECT + 5);
+		byte[] data3 = rng.nextBytes(ObjectLoader.STREAM_THRESHOLD + 5);
 		ByteArrayOutputStream tmp = new ByteArrayOutputStream();
 		DeltaEncoder de = new DeltaEncoder(tmp, data0.length, data3.length);
 		de.insert(data3, 0, data3.length);
 		byte[] delta3 = tmp.toByteArray();
-		assertTrue(delta3.length > UnpackedObject.LARGE_OBJECT);
+		assertTrue(delta3.length > ObjectLoader.STREAM_THRESHOLD);
 
 		TemporaryBuffer.Heap pack = new TemporaryBuffer.Heap(64 * 1024);
 		packHeader(pack, 2);

org.eclipse.jgit.test/tst/org/eclipse/jgit/storage/file/UnpackedObjectTest.java

@@ -113,7 +113,7 @@ public class UnpackedObjectTest extends LocalDiskRepositoryTestCase {
 
 	public void testStandardFormat_LargeObject() throws Exception {
 		final int type = Constants.OBJ_BLOB;
-		byte[] data = rng.nextBytes(UnpackedObject.LARGE_OBJECT + 5);
+		byte[] data = rng.nextBytes(ObjectLoader.STREAM_THRESHOLD + 5);
 		ObjectId id = new ObjectInserter.Formatter().idFor(type, data);
 		write(id, compressStandardFormat(type, data));
 
@@ -230,7 +230,7 @@ public class UnpackedObjectTest extends LocalDiskRepositoryTestCase {
 	public void testStandardFormat_LargeObject_CorruptZLibStream()
 			throws Exception {
 		final int type = Constants.OBJ_BLOB;
-		byte[] data = rng.nextBytes(UnpackedObject.LARGE_OBJECT + 5);
+		byte[] data = rng.nextBytes(ObjectLoader.STREAM_THRESHOLD + 5);
 		ObjectId id = new ObjectInserter.Formatter().idFor(type, data);
 		byte[] gz = compressStandardFormat(type, data);
 		gz[gz.length - 1] = 0;
@@ -290,7 +290,7 @@ public class UnpackedObjectTest extends LocalDiskRepositoryTestCase {
 
 	public void testPackFormat_LargeObject() throws Exception {
 		final int type = Constants.OBJ_BLOB;
-		byte[] data = rng.nextBytes(UnpackedObject.LARGE_OBJECT + 5);
+		byte[] data = rng.nextBytes(ObjectLoader.STREAM_THRESHOLD + 5);
 		ObjectId id = new ObjectInserter.Formatter().idFor(type, data);
 		write(id, compressPackFormat(type, data));
 

org.eclipse.jgit/src/org/eclipse/jgit/events/ConfigChangedEvent.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (C) 2010, Google Inc.
+ * and other copyright owners as documented in the project's IP log.
+ *
+ * This program and the accompanying materials are made available
+ * under the terms of the Eclipse Distribution License v1.0 which
+ * accompanies this distribution, is reproduced below, and is
+ * available at http://www.eclipse.org/org/documents/edl-v10.php
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following
+ *   disclaimer in the documentation and/or other materials provided
+ *   with the distribution.
+ *
+ * - Neither the name of the Eclipse Foundation, Inc. nor the
+ *   names of its contributors may be used to endorse or promote
+ *   products derived from this software without specific prior
+ *   written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
+ * CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+ * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
+ * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+ * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
+ * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package org.eclipse.jgit.events;
+
+/** Describes a change to one or more keys in the configuration. */
+public class ConfigChangedEvent extends RepositoryEvent<ConfigChangedListener> {
+	@Override
+	public Class<ConfigChangedListener> getListenerType() {
+		return ConfigChangedListener.class;
+	}
+
+	@Override
+	public void dispatch(ConfigChangedListener listener) {
+		listener.onConfigChanged(this);
+	}
+}

org.eclipse.jgit/src/org/eclipse/jgit/events/ConfigChangedListener.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright (C) 2008, Robin Rosenberg <robin.rosenberg@dewire.com>
+ * and other copyright owners as documented in the project's IP log.
+ *
+ * This program and the accompanying materials are made available
+ * under the terms of the Eclipse Distribution License v1.0 which
+ * accompanies this distribution, is reproduced below, and is
+ * available at http://www.eclipse.org/org/documents/edl-v10.php
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or
+ * without modification, are permitted provided that the following
+ * conditions are met:
+ *
+ * - Redistributions of source code must retain the above copyright
+ *   notice, this list of conditions and the following disclaimer.
+ *
+ * - Redistributions in binary form must reproduce the above
+ *   copyright notice, this list of conditions and the following
+ *   disclaimer in the documentation and/or other materials provided
+ *   with the distribution.
+ *
+ * - Neither the name of the Eclipse Foundation, Inc. nor the
+ *   names of its contributors may be used to endorse or promote
+ *   products derived from this software without specific prior
+ *   written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
+ * CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+ * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+ * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
+ * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+ * STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
+ * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package org.eclipse.jgit.events;
+
+/** Receives {@link ConfigChangedEvent}s. */
+public interface ConfigChangedListener extends RepositoryListener {
+	/**
+	 * Invoked when any change is made to the configuration.
+	 *
+	 * @param event
+	 *            information about the changes.
+	 */
+	void onConfigChanged(ConfigChangedEvent event);
+}

org.eclipse.jgit/src/org/eclipse/jgit/events/ListenerList.java

@@ -74,6 +74,18 @@ public class ListenerList {
 		return addListener(RefsChangedListener.class, listener);
 	}
 
+	/**
+	 * Register a ConfigChangedListener.
+	 *
+	 * @param listener
+	 *            the listener implementation.
+	 * @return handle to later remove the listener.
+	 */
+	public ListenerHandle addConfigChangedListener(
+			ConfigChangedListener listener) {
+		return addListener(ConfigChangedListener.class, listener);
+	}
+
 	/**
 	 * Add a listener to the list.
 	 *

org.eclipse.jgit/src/org/eclipse/jgit/lib/CoreConfig.java

@@ -47,6 +47,7 @@
 package org.eclipse.jgit.lib;
 
 import static java.util.zip.Deflater.DEFAULT_COMPRESSION;
+import static org.eclipse.jgit.lib.ObjectLoader.STREAM_THRESHOLD;
 
 import org.eclipse.jgit.lib.Config.SectionParser;
 
@@ -67,10 +68,18 @@ public class CoreConfig {
 
 	private final boolean logAllRefUpdates;
 
+	private final int streamFileThreshold;
+
 	private CoreConfig(final Config rc) {
 		compression = rc.getInt("core", "compression", DEFAULT_COMPRESSION);
 		packIndexVersion = rc.getInt("pack", "indexversion", 2);
 		logAllRefUpdates = rc.getBoolean("core", "logallrefupdates", true);
+
+		long maxMem = Runtime.getRuntime().maxMemory();
+		long sft = rc.getLong("core", null, "streamfilethreshold", STREAM_THRESHOLD);
+		sft = Math.min(sft, maxMem / 4); // don't use more than 1/4 of the heap
+		sft = Math.min(sft, Integer.MAX_VALUE); // cannot exceed array length
+		streamFileThreshold = (int) sft;
 	}
 
 	/**
@@ -94,4 +103,9 @@ public class CoreConfig {
 	public boolean isLogAllRefUpdates() {
 		return logAllRefUpdates;
 	}
+
+	/** @return the size threshold beyond which objects must be streamed. */
+	public int getStreamFileThreshold() {
+		return streamFileThreshold;
+	}
 }

org.eclipse.jgit/src/org/eclipse/jgit/lib/ObjectLoader.java

@@ -59,7 +59,13 @@ import org.eclipse.jgit.errors.MissingObjectException;
  * New loaders are constructed for every object.
  */
 public abstract class ObjectLoader {
-	private static final int LARGE_OBJECT = 1024 * 1024;
+	/**
+	 * Default setting for the large object threshold.
+	 * <p>
+	 * Objects larger than this size must be accessed as a stream through the
+	 * loader's {@link #openStream()} method.
+	 */
+	public static final int STREAM_THRESHOLD = 1024 * 1024;
 
 	/**
 	 * @return Git in pack object type, see {@link Constants}.
@@ -77,7 +83,12 @@ public abstract class ObjectLoader {
 	 *         {@link #openStream()} to prevent overflowing the JVM heap.
 	 */
 	public boolean isLarge() {
-		return LARGE_OBJECT <= getSize();
+		try {
+			getCachedBytes();
+			return false;
+		} catch (LargeObjectException tooBig) {
+			return true;
+		}
 	}
 
 	/**

org.eclipse.jgit/src/org/eclipse/jgit/storage/file/CachedObjectDirectory.java

@@ -191,4 +191,9 @@ class CachedObjectDirectory extends FileObjectDatabase {
 			WindowCursor curs) throws IOException {
 		wrapped.selectObjectRepresentation(packer, otp, curs);
 	}
+
+	@Override
+	int getStreamFileThreshold() {
+		return wrapped.getStreamFileThreshold();
+	}
 }

org.eclipse.jgit/src/org/eclipse/jgit/storage/file/FileObjectDatabase.java

@@ -187,6 +187,8 @@ abstract class FileObjectDatabase extends ObjectDatabase {
 
 	abstract FileObjectDatabase newCachedFileObjectDatabase();
 
+	abstract int getStreamFileThreshold();
+
 	static class AlternateHandle {
 		final FileObjectDatabase db;
 

org.eclipse.jgit/src/org/eclipse/jgit/storage/file/FileRepository.java

@@ -146,6 +146,7 @@ public class FileRepository extends Repository {
 				options.getObjectDirectory(), //
 				options.getAlternateObjectDirectories(), //
 				getFS());
+		getListenerList().addConfigChangedListener(objectDatabase);
 
 		if (objectDatabase.exists()) {
 			final String repositoryFormatVersion = getConfig().getString(

org.eclipse.jgit/src/org/eclipse/jgit/storage/file/LargePackedDeltaObject.java

@@ -93,6 +93,11 @@ class LargePackedDeltaObject extends ObjectLoader {
 		return size;
 	}
 
+	@Override
+	public boolean isLarge() {
+		return true;
+	}
+
 	@Override
 	public byte[] getCachedBytes() throws LargeObjectException {
 		try {

org.eclipse.jgit/src/org/eclipse/jgit/storage/file/ObjectDirectory.java

@@ -63,8 +63,11 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import org.eclipse.jgit.JGitText;
 import org.eclipse.jgit.errors.PackMismatchException;
+import org.eclipse.jgit.events.ConfigChangedEvent;
+import org.eclipse.jgit.events.ConfigChangedListener;
 import org.eclipse.jgit.lib.AnyObjectId;
 import org.eclipse.jgit.lib.Config;
+import org.eclipse.jgit.lib.CoreConfig;
 import org.eclipse.jgit.lib.ObjectDatabase;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.lib.ObjectInserter;
@@ -93,7 +96,8 @@ import org.eclipse.jgit.util.FS;
  * searched (recursively through all alternates) before the slow half is
  * considered.
  */
-public class ObjectDirectory extends FileObjectDatabase {
+public class ObjectDirectory extends FileObjectDatabase implements
+		ConfigChangedListener {
 	private static final PackList NO_PACKS = new PackList(-1, -1, new PackFile[0]);
 
 	private final Config config;
@@ -112,6 +116,8 @@ public class ObjectDirectory extends FileObjectDatabase {
 
 	private final AtomicReference<AlternateHandle[]> alternates;
 
+	private int streamFileThreshold;
+
 	/**
 	 * Initialize a reference to an on-disk object directory.
 	 *
@@ -146,6 +152,13 @@ public class ObjectDirectory extends FileObjectDatabase {
 				alt[i] = openAlternate(alternatePaths[i]);
 			alternates.set(alt);
 		}
+
+		onConfigChanged(new ConfigChangedEvent());
+	}
+
+	public void onConfigChanged(ConfigChangedEvent event) {
+		CoreConfig core = config.get(CoreConfig.KEY);
+		streamFileThreshold = core.getStreamFileThreshold();
 	}
 
 	/**
@@ -632,4 +645,9 @@ public class ObjectDirectory extends FileObjectDatabase {
 	FileObjectDatabase newCachedFileObjectDatabase() {
 		return new CachedObjectDirectory(this);
 	}
+
+	@Override
+	int getStreamFileThreshold() {
+		return streamFileThreshold;
+	}
 }

org.eclipse.jgit/src/org/eclipse/jgit/storage/file/PackFile.java

@@ -632,7 +632,7 @@ public class PackFile implements Iterable<PackIndex.MutableEntry> {
 			case Constants.OBJ_TREE:
 			case Constants.OBJ_BLOB:
 			case Constants.OBJ_TAG: {
-				if (sz < UnpackedObject.LARGE_OBJECT) {
+				if (sz < curs.getStreamFileThreshold()) {
 					byte[] data = decompress(pos + p, sz, curs);
 					return new ObjectLoader.SmallObject(type, data);
 				}
@@ -683,7 +683,7 @@ public class PackFile implements Iterable<PackIndex.MutableEntry> {
 	private ObjectLoader loadDelta(long posSelf, int hdrLen, long sz,
 			long posBase, WindowCursor curs) throws IOException,
 			DataFormatException {
-		if (UnpackedObject.LARGE_OBJECT <= sz) {
+		if (curs.getStreamFileThreshold() <= sz) {
 			// The delta instruction stream itself is pretty big, and
 			// that implies the resulting object is going to be massive.
 			// Use only the large delta format here.

org.eclipse.jgit/src/org/eclipse/jgit/storage/file/UnpackedObject.java

@@ -76,8 +76,6 @@ import org.eclipse.jgit.util.RawParseUtils;
 public class UnpackedObject {
 	private static final int BUFFER_SIZE = 8192;
 
-	static final int LARGE_OBJECT = 1024 * 1024;
-
 	/**
 	 * Parse an object from the unpacked object format.
 	 *
@@ -127,7 +125,7 @@ public class UnpackedObject {
 							JGitText.get().corruptObjectGarbageAfterSize);
 				if (path == null && Integer.MAX_VALUE < size)
 					throw new LargeObjectException(id.copy());
-				if (size < LARGE_OBJECT || path == null) {
+				if (size < wc.getStreamFileThreshold() || path == null) {
 					byte[] data = new byte[(int) size];
 					int n = avail - p.value;
 					if (n > 0)
@@ -164,7 +162,7 @@ public class UnpackedObject {
 
 				if (path == null && Integer.MAX_VALUE < size)
 					throw new LargeObjectException(id.copy());
-				if (size < LARGE_OBJECT || path == null) {
+				if (size < wc.getStreamFileThreshold() || path == null) {
 					in.reset();
 					IO.skipFully(in, p);
 					in = inflate(in, wc);

org.eclipse.jgit/src/org/eclipse/jgit/storage/file/WindowCursor.java

@@ -223,6 +223,12 @@ final class WindowCursor extends ObjectReader implements ObjectReuseAsIs {
 		}
 	}
 
+	int getStreamFileThreshold() {
+		if (db == null)
+			return ObjectLoader.STREAM_THRESHOLD;
+		return db.getStreamFileThreshold();
+	}
+
 	/** Release the current window cursor. */
 	public void release() {
 		window = null;