[teiid-commits] teiid SVN: r3531 - trunk/engine/src/main/java/org/teiid/common/buffer/impl.

[teiid-commits at lists.jboss.org]

Author: shawkins
Date: 2011-10-05 12:09:13 -0400 (Wed, 05 Oct 2011)
New Revision: 3531

Modified:
   trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java
   trunk/engine/src/main/java/org/teiid/common/buffer/impl/OrderedCache.java
Log:
TEIID-1750 changing memoryentries to be non-blocking

Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java	2011-10-05 16:05:38 UTC (rev 3530)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/BufferManagerImpl.java	2011-10-05 16:09:13 UTC (rev 3531)
@@ -296,8 +296,6 @@
 			value.setOrderingValue(orderingValue);
 		}
 	};
-	
-	//private LinkedHashMap<Long, CacheEntry> memoryEntries = new LinkedHashMap<Long, CacheEntry>();
     
     //limited size reference caches based upon the memory settings
     private WeakReferenceHashedValueCache<CacheEntry> weakReferenceCache; 
@@ -554,14 +552,13 @@
 		int maxToFree = Math.max(maxProcessingKB>>1, reserveBatchKB>>3);
 		int freed = 0;
 		while (true) {
-			CacheEntry ce = null;
-			synchronized (memoryEntries) {
-				if (freed > maxToFree || activeBatchKB.get() == 0 || activeBatchKB.get() < reserveBatchKB * .8) {
-					break;
-				}
-				ce = memoryEntries.evict();
-				freed += ce.getSizeEstimate();
-				activeBatchKB.addAndGet(-ce.getSizeEstimate());
+			if (freed > maxToFree || activeBatchKB.get() == 0 || activeBatchKB.get() < reserveBatchKB * .8) {
+				break;
+			}
+			CacheEntry ce = memoryEntries.evict();
+			freed += ce.getSizeEstimate();
+			activeBatchKB.addAndGet(-ce.getSizeEstimate());
+			synchronized (ce) {
 				if (ce.isPersistent()) {
 					continue;
 				}
@@ -603,12 +600,10 @@
 	 */
 	CacheEntry fastGet(Long batch, boolean prefersMemory, boolean retain) {
 		CacheEntry ce = null;
-		synchronized (memoryEntries) {
-			if (retain) {
-				ce = memoryEntries.get(batch);
-			} else {
-				ce = memoryEntries.remove(batch);
-			}
+		if (retain) {
+			ce = memoryEntries.get(batch);
+		} else {
+			ce = memoryEntries.remove(batch);
 		}
 		if (ce != null) {
 			if (!retain) {
@@ -667,7 +662,7 @@
 	
 	void addMemoryEntry(CacheEntry ce, CacheEntry previous) {
 		persistBatchReferences();
-		synchronized (memoryEntries) {
+		synchronized (ce) {
 			if (previous != null) {
 				ce.setOrderingValue(previous.getOrderingValue());
 			}

Modified: trunk/engine/src/main/java/org/teiid/common/buffer/impl/OrderedCache.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/common/buffer/impl/OrderedCache.java	2011-10-05 16:05:38 UTC (rev 3530)
+++ trunk/engine/src/main/java/org/teiid/common/buffer/impl/OrderedCache.java	2011-10-05 16:09:13 UTC (rev 3531)
@@ -22,30 +22,24 @@
 
 package org.teiid.common.buffer.impl;
 
-import java.util.Comparator;
-import java.util.HashMap;
 import java.util.Map;
-import java.util.TreeMap;
+import java.util.NavigableMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
 
 public abstract class OrderedCache<K, V> {
 	
-	protected HashMap<K, V> map = new HashMap<K, V>(); 
-	protected TreeMap<V, K> expirationQueue;
-	
-	public OrderedCache() {
-		expirationQueue = new TreeMap<V, K>();
-	}
-	
-	public OrderedCache(Comparator<? super V> comparator) {
-		expirationQueue = new TreeMap<V, K>(comparator);
-	}
-	
+	protected Map<K, V> map = new ConcurrentHashMap<K, V>(); 
+	protected NavigableMap<V, K> expirationQueue = new ConcurrentSkipListMap<V, K>();
+		
 	public V get(K key) {
 		V result = map.get(key);
 		if (result != null) {
-			expirationQueue.remove(result);
-			recordAccess(key, result, false);
-			expirationQueue.put(result, key);
+			synchronized (result) {
+				expirationQueue.remove(result);
+				recordAccess(key, result, false);
+				expirationQueue.put(result, key);
+			}
 		}
 		return result;
 	}
@@ -53,7 +47,9 @@
 	public V remove(K key) {
 		V result = map.remove(key);
 		if (result != null) {
-			expirationQueue.remove(result);
+			synchronized (result) {
+				expirationQueue.remove(result);
+			}
 		}
 		return result;
 	}
@@ -61,10 +57,14 @@
 	public V put(K key, V value) {
 		V result = map.put(key, value);
 		if (result != null) {
-			expirationQueue.remove(result);
+			synchronized (result) {
+				expirationQueue.remove(result);
+			}
 		}
-		recordAccess(key, value, result == null);
-		expirationQueue.put(value, key);
+		synchronized (value) {
+			recordAccess(key, value, result == null);
+			expirationQueue.put(value, key);
+		}
 		return result;
 	}
 	
@@ -73,8 +73,7 @@
 		if (entry == null) {
 			return null;
 		}
-		map.remove(entry.getValue());
-		return entry.getKey();
+		return map.remove(entry.getValue());
 	}
 	
 	public int size() {