fixed #52

Author: mgeipel

src/main/java/org/culturegraph/mf/stream/pipe/sort/AbstractTripleSort.java

@@ -29,31 +29,33 @@
 import org.culturegraph.mf.framework.DefaultObjectPipe;
 import org.culturegraph.mf.framework.ObjectReceiver;
 import org.culturegraph.mf.types.Triple;
-
+import org.culturegraph.mf.util.MemoryWarningSystem;
+import org.culturegraph.mf.util.MemoryWarningSystem.Listener;
 
 /**
  * @author markus geipel
- *
+ * 
  */
-public abstract class AbstractTripleSort extends DefaultObjectPipe<Triple, ObjectReceiver<Triple>> {
+public abstract class AbstractTripleSort extends DefaultObjectPipe<Triple, ObjectReceiver<Triple>> implements Listener {
 	/**
 	 * specifies the comparator
 	 */
 	public enum Compare {
 		SUBJECT, PREDICATE, OBJECT, ALL;
 	}
-	
+
 	/**
 	 * sort order
-	 *
+	 * 
 	 */
 	public enum Order {
 		INCREASING {
 			@Override
 			public int order(final int indicator) {
 				return indicator;
 			}
-		}, DECREASING {
+		},
+		DECREASING {
 			@Override
 			public int order(final int indicator) {
 				return -indicator;
@@ -62,25 +64,22 @@ public int order(final int indicator) {
 		public abstract int order(int indicator);
 	}
 
-
-	private static final int KILO = 1024;
-	private static final int DEFUALT_BLOCKSIZE = 128 * KILO * KILO;
-	private static final int STRING_OVERHEAD = 124;
-	
 	private final List<Triple> buffer = new ArrayList<Triple>();
-	private final List<File> tempFiles = new ArrayList<File>();
+	private final List<File> tempFiles;
 	private Compare compare = Compare.SUBJECT;
 	private Order order = Order.INCREASING;
-	//private Comparator<Triple> comparator = createComparator(compareBy, order);
-	private long bufferSizeEstimate;
+	private volatile boolean memoryLow;
 
-	private long blockSize = DEFUALT_BLOCKSIZE;
+	public AbstractTripleSort() {
+		MemoryWarningSystem.addListener(this);
+		tempFiles = new ArrayList<File>(); // Initialized here to let the
+											// compiler enforce the call to
+											// super() in subclasses.
+	}
 
-	/**
-	 * @param blockSize in MB
-	 */
-	public final void setBlockSize(final int blockSize) {
-		this.blockSize = blockSize * KILO * KILO;
+	@Override
+	public final void memoryLow(final long usedMemory, final long maxMemory) {
+		memoryLow = true;
 	}
 
 	protected final void setCompare(final Compare compare) {
@@ -90,29 +89,23 @@ protected final void setCompare(final Compare compare) {
 	protected final Compare getCompare() {
 		return compare;
 	}
-	
-	protected final void setSortOrder(final Order order){
+
+	protected final void setSortOrder(final Order order) {
 		this.order = order;
 	}
 
-
-
 	@Override
 	public final void process(final Triple namedValue) {
-
-		buffer.add(namedValue);
-		// padding is ignored for efficiency (overhead is 45 for name + 45 for
-		// value + 8 for namedValue + 28 goodwill)
-		bufferSizeEstimate += ((namedValue.getSubject().length() + namedValue.getPredicate().length() + namedValue.getObject().length()) * 2) + STRING_OVERHEAD;
-		if (bufferSizeEstimate > blockSize) {
-			bufferSizeEstimate = 0;
+		if (memoryLow) {
 			try {
 				nextBatch();
 			} catch (IOException e) {
 				throw new MetafactureException("Error writing to temp file after sorting", e);
+			} finally {
+				memoryLow = false;
 			}
-
 		}
+		buffer.add(namedValue);
 	}
 
 	private void nextBatch() throws IOException {
@@ -132,11 +125,9 @@ private void nextBatch() throws IOException {
 		tempFiles.add(tempFile);
 	}
 
-
 	@Override
 	public final void onCloseStream() {
 
-
 		if (tempFiles.isEmpty()) {
 			Collections.sort(buffer, createComparator(compare, order));
 			for (Triple triple : buffer) {
@@ -147,7 +138,8 @@ public final void onCloseStream() {
 			final Comparator<Triple> comparator = createComparator(compare, order);
 			final PriorityQueue<SortedTripleFileFacade> queue = new PriorityQueue<SortedTripleFileFacade>(11,
 					new Comparator<SortedTripleFileFacade>() {
-					//	private final Comparator<Triple> comparator = getComparator();
+						// private final Comparator<Triple> comparator =
+						// getComparator();
 
 						@Override
 						public int compare(final SortedTripleFileFacade o1, final SortedTripleFileFacade o2) {
@@ -183,15 +175,15 @@ public int compare(final SortedTripleFileFacade o1, final SortedTripleFileFacade
 
 	protected void onFinished() {
 		// nothing to do
-		
+
 	}
 
 	protected abstract void sortedTriple(Triple namedValue);
 
-	public final Comparator<Triple> createComparator(){
+	public final Comparator<Triple> createComparator() {
 		return createComparator(compare, order);
 	}
-	
+
 	public static Comparator<Triple> createComparator(final Compare compareBy, final Order order) {
 		final Comparator<Triple> comparator;
 		switch (compareBy) {
@@ -236,8 +228,8 @@ public int compare(final Triple o1, final Triple o2) {
 	@Override
 	public final void onResetStream() {
 		buffer.clear();
-		for(File file: tempFiles){
-			if(file.exists()){
+		for (File file : tempFiles) {
+			if (file.exists()) {
 				file.delete();
 			}
 		}

src/main/java/org/culturegraph/mf/util/MemoryWarningSystem.java

@@ -0,0 +1,113 @@
+/*
+ *  Copyright 2013 Deutsche Nationalbibliothek
+ *
+ *  Licensed under the Apache License, Version 2.0 the "License";
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+/*
+ *  Code based on http://www.javaspecialists.eu/archive/Issue092.html 
+ */
+
+package org.culturegraph.mf.util;
+
+import java.lang.management.ManagementFactory;
+import java.lang.management.MemoryMXBean;
+import java.lang.management.MemoryNotificationInfo;
+import java.lang.management.MemoryPoolMXBean;
+import java.lang.management.MemoryType;
+import java.util.Collection;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import javax.management.Notification;
+import javax.management.NotificationEmitter;
+import javax.management.NotificationListener;
+
+/**
+ * This memory warning system will call the listener when we exceed the
+ * percentage of available memory specified. The class is static, since the usage threshold can only be set to one
+ * number.
+ */
+public final class MemoryWarningSystem {
+	private static final MemoryPoolMXBean TENURED_GEN_POOL = findTenuredGenPool();
+	private static final double DEFAULT_THRESHOLD = 0.8;
+	private static final Collection<Listener> LISTENERS = new CopyOnWriteArrayList<Listener>();
+
+	private MemoryWarningSystem() {
+		// no instances
+	}
+	
+	static {
+		setUsageThreshold(DEFAULT_THRESHOLD);
+		final MemoryMXBean mbean = ManagementFactory.getMemoryMXBean();
+		final NotificationEmitter emitter = (NotificationEmitter) mbean;
+		emitter.addNotificationListener(new NotificationListener() {
+			
+			@Override
+			public void handleNotification(final Notification notification, final Object handback) {
+				if (notification.getType().equals(MemoryNotificationInfo.MEMORY_THRESHOLD_EXCEEDED)) {
+					for (Listener listener : getListeners()) {
+						listener.memoryLow(getUsedMemory(), getMaxMemory());
+					}
+				}
+			}
+		}, null, null);
+	}
+
+	protected static Collection<Listener> getListeners(){
+		return LISTENERS;
+	}
+	
+	public static long getMaxMemory() {
+		return TENURED_GEN_POOL.getUsage().getMax();
+	}
+
+	public static long getUsedMemory() {
+		return TENURED_GEN_POOL.getUsage().getUsed();
+	}
+
+	public static boolean addListener(final Listener listener) {
+		return LISTENERS.add(listener);
+	}
+
+	public static boolean removeListener(final Listener listener) {
+		return LISTENERS.remove(listener);
+	}
+
+	public static void setUsageThreshold(final double threshold) {
+		if (threshold <= 0.0 || threshold > 1.0) {
+			throw new IllegalArgumentException("'threshold' must be in [0.0, 1.0]");
+		}
+		TENURED_GEN_POOL.setUsageThreshold((long) (getMaxMemory() * threshold));
+	}
+
+	/**
+	 * Tenured Space Pool can be determined by it being of type HEAP and by it
+	 * being possible to set the usage threshold.
+	 */
+	private static MemoryPoolMXBean findTenuredGenPool() {
+		for (MemoryPoolMXBean pool : ManagementFactory.getMemoryPoolMXBeans()) {
+			// I don't know whether this approach is better, or whether
+			// we should rather check for the pool name "Tenured Gen"?
+			if (pool.getType() == MemoryType.HEAP && pool.isUsageThresholdSupported()) {
+				return pool;
+			}
+		}
+		throw new AssertionError("Could not find tenured space");
+	}
+	
+	/**
+	 * Interface for low memory listeners
+	 */
+	public interface Listener {
+		void memoryLow(long usedMemory, long maxMemory);
+	}
+}