Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/XAWorkspace.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/XAWorkspace.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/XAWorkspace.java	(working copy)
@@ -53,7 +53,7 @@
      * {@inheritDoc}
      */
     protected LocalItemStateManager createItemStateManager(SharedItemStateManager shared) {
-        return new XAItemStateManager(shared, this);
+        return new XAItemStateManager(shared, this, rep.getItemStateCacheFactory());
     }
 
     /**
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/XASessionImpl.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/XASessionImpl.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/XASessionImpl.java	(working copy)
@@ -131,7 +131,7 @@
          * Create array that contains all resources that paricipate in this
          * transactions. Because some resources depend on each other, there is
          * also a workspace scoped lock resource inserted, that guards the
-         * entire transaction from deadlocks (see JCR-335) 
+         * entire transaction from deadlocks (see JCR-335)
          */
         txResources = new InternalXAResource[] {
             ((XAWorkspace) wsp).getXAResourceBegin(),
@@ -158,7 +158,7 @@
             throws RepositoryException {
 
         VersionManagerImpl vMgr = (VersionManagerImpl) rep.getVersionManager();
-        return new XAVersionManager(vMgr, rep.getNodeTypeRegistry(), this);
+        return new XAVersionManager(vMgr, rep.getNodeTypeRegistry(), this, rep.getItemStateCacheFactory());
     }
 
     /**
@@ -373,6 +373,19 @@
     }
 
     /**
+     * {@inheritDoc}
+     */
+    public synchronized void logout() {
+        super.logout();
+        // dispose the caches
+        try {
+            ((XAVersionManager)versionMgr).close();
+        } catch(Exception e) {
+            log.warn("error while closing XAVersionManager", e);
+        }
+    }
+
+    /**
      * Compare two strings for equality. If both are <code>null</code>, this
      * is also considered to be equal.
      */
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateReferenceCache.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateReferenceCache.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateReferenceCache.java	(working copy)
@@ -66,8 +66,8 @@
      * <code>LRUItemStateCache</code> instance as internal secondary
      * cache.
      */
-    public ItemStateReferenceCache() {
-        this(new MLRUItemStateCache());
+    public ItemStateReferenceCache(ItemStateCacheFactory cacheFactory) {
+        this(cacheFactory.newItemStateCache());
     }
 
     /**
@@ -130,6 +130,13 @@
     /**
      * {@inheritDoc}
      */
+    public void dispose() {
+        cache.dispose();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
     public void evictAll() {
         // fake call to update stats of secondary cache
         cache.evictAll();
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/Cache.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/Cache.java	(revision 0)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/Cache.java	(revision 0)
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.jackrabbit.core.state;
+
+/**
+ * A <code>Cache</code> object
+ * A cache must call <code>CacheManager.getInstance().add(this)</code>
+ * to take part in the dynamic memory distribution.
+ */
+public interface Cache {
+
+    /**
+     * Set the new memory limit.
+     * @param size the size in bytes
+     */
+    void setMaxMemorySize(long size);
+
+    /**
+     * Get the current limit.
+     * @return the size in bytes
+     */
+    long getMaxMemorySize();
+
+    /**
+     * Get the amount of used memory.
+     * @return the size in bytes
+     */
+    long getMemoryUsed();
+
+    /**
+     * Get the number of accesses (get or set) until resetAccessCount was called.
+     * @return the count
+     */
+
+    long getAccessCount();
+
+    /**
+     * Reset the access counter.
+     */
+    void resetAccessCount();
+
+    /**
+     * Add a listener to this cache that is informed after a number of accesses.
+     */
+    void setAccessListener(CacheAccessListener listener);
+
+}
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/LocalItemStateManager.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/LocalItemStateManager.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/LocalItemStateManager.java	(working copy)
@@ -70,8 +70,8 @@
      * @param factory event state collection factory
      */
     public LocalItemStateManager(SharedItemStateManager sharedStateMgr,
-                                 EventStateCollectionFactory factory) {
-        cache = new ItemStateReferenceCache();
+                                 EventStateCollectionFactory factory, ItemStateCacheFactory cacheFactory) {
+        cache = new ItemStateReferenceCache(cacheFactory);
         this.sharedStateMgr = sharedStateMgr;
         this.factory = factory;
 
@@ -344,6 +344,7 @@
         }
         // clear cache
         cache.evictAll();
+        cache.dispose();
     }
 
     /**
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/LRUItemStateCache.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/LRUItemStateCache.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/LRUItemStateCache.java	(working copy)
@@ -134,4 +134,11 @@
     public Collection values() {
         return Collections.unmodifiableCollection(cache.values());
     }
+
+    /**
+     * {@inheritDoc}
+     */
+    public void dispose() {
+        cache.clear();
+    }
 }
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/SharedItemStateManager.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/SharedItemStateManager.java	(revision 474304)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/SharedItemStateManager.java	(working copy)
@@ -196,9 +196,10 @@
     public SharedItemStateManager(PersistenceManager persistMgr,
                                   NodeId rootNodeId,
                                   NodeTypeRegistry ntReg,
-                                  boolean usesReferences)
+                                  boolean usesReferences,
+                                  ItemStateCacheFactory cacheFactory)
             throws ItemStateException {
-        cache = new ItemStateReferenceCache();
+        cache = new ItemStateReferenceCache(cacheFactory);
         this.persistMgr = persistMgr;
         this.ntReg = ntReg;
         this.usesReferences = usesReferences;
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/MLRUItemStateCache.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/MLRUItemStateCache.java	(revision 474304)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/MLRUItemStateCache.java	(working copy)
@@ -34,7 +34,7 @@
  * cache uses a rough estimate of the memory consuption of the cache item
  * states for calculating the maximum number of entries.
  */
-public class MLRUItemStateCache implements ItemStateCache {
+public class MLRUItemStateCache implements ItemStateCache, Cache {
     /** Logger instance */
     private static Logger log = LoggerFactory.getLogger(LRUItemStateCache.class);
 
@@ -45,11 +45,17 @@
     private long totalMem;
 
     /** the maximum of memory the cache may use */
-    private final long maxMem;
+    private long maxMem;
 
     /** the number of writes */
     private long numWrites = 0;
 
+    /** the access count */
+    private long accessCount = 0;
+
+    /** the cache access listeners */
+    private CacheAccessListener accessListener;
+
     /**
      * A cache for <code>ItemState</code> instances
      */
@@ -69,7 +75,7 @@
      *
      * @param maxMem the maximum amount of memory this cache may use.
      */
-    public MLRUItemStateCache(int maxMem) {
+    private MLRUItemStateCache(int maxMem) {
         this.maxMem = maxMem;
     }
 
@@ -88,6 +94,7 @@
      */
     public ItemState retrieve(ItemId id) {
         synchronized (cache) {
+            touch();
             Entry entry = (Entry) cache.remove(id);
             if (entry != null) {
                 // 'touch' item, by adding at end of list
@@ -104,6 +111,7 @@
      */
     public void update(ItemId id) {
         synchronized (cache) {
+            touch();
             Entry entry = (Entry) cache.get(id);
             if (entry != null) {
                 totalMem -= entry.size;
@@ -118,6 +126,7 @@
      */
     public void cache(ItemState state) {
         synchronized (cache) {
+            touch();
             ItemId id = state.getId();
             if (cache.containsKey(id)) {
                 log.warn("overwriting cached entry " + id);
@@ -126,22 +135,27 @@
             Entry entry = new Entry(state);
             cache.put(id, entry);
             totalMem += entry.size;
-            // remove items, if too many
-            while (totalMem > maxMem) {
-                id = (ItemId) cache.firstKey();
-                evict(id);
-            }
+            shrinkIfRequired();
             if (numWrites++%10000 == 0 && log.isDebugEnabled()) {
                 log.info(this + " size=" + cache.size() + ", " + totalMem + "/" + maxMem);
             }
         }
     }
 
+    private void shrinkIfRequired() {
+        // remove items, if too many
+        while (totalMem > maxMem) {
+            ItemId id = (ItemId) cache.firstKey();
+            evict(id);
+        }
+    }
+
     /**
      * {@inheritDoc}
      */
     public void evict(ItemId id) {
         synchronized (cache) {
+            touch();
             Entry entry = (Entry) cache.remove(id);
             if (entry != null) {
                 totalMem -= entry.size;
@@ -201,9 +215,88 @@
         }
     }
 
+    private void touch() {
+        accessCount++;
+        if ((accessCount % CacheAccessListener.ACCESS_INTERVAL) == 0) {
+            if (accessListener != null) {
+                accessListener.cacheAccessed();
+            }
+        }
+    }
+
     /**
-     * Internal cache entry
+     * {@inheritDoc}
      */
+    public long getAccessCount() {
+        return accessCount;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public long getMaxMemorySize() {
+        return maxMem;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public long getMemoryUsed() {
+        synchronized (cache) {
+            totalMem = 0;
+            Iterator iter = cache.values().iterator();
+            while (iter.hasNext()) {
+                Entry entry = (Entry) iter.next();
+                entry.recalc();
+                totalMem += entry.size;
+            }
+        }
+        return totalMem;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public void resetAccessCount() {
+        synchronized (cache) {
+            accessCount = 0;
+        }
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public void setMaxMemorySize(long size) {
+        synchronized (cache) {
+            this.maxMem = size;
+            shrinkIfRequired();
+        }
+    }
+
+    /**
+     * Set the cache access listener. Only one listener per cache is supported.
+     *
+     * @param listener the new listener
+     */
+    public void setAccessListener(CacheAccessListener listener) {
+        this.accessListener = listener;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public void dispose() {
+        synchronized (cache) {
+            if(accessListener != null) {
+                accessListener.disposeCache(this);
+            }
+        }
+    }
+
+
+    /**
+     * Internal cache entry.
+     */
     private static class Entry {
 
         private final ItemState state;
@@ -219,4 +312,5 @@
             size = 64 + state.calculateMemoryFootprint();
         }
     }
+
 }
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ManagedMLRUItemStateCacheFactory.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ManagedMLRUItemStateCacheFactory.java	(revision 0)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ManagedMLRUItemStateCacheFactory.java	(revision 0)
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.jackrabbit.core.state;
+
+/**
+ * This class constructs new MLRUItemStateCache.
+ * This class adds the new caches to the cache manager,
+ * and links the caches to the cache manager.
+ */
+public class ManagedMLRUItemStateCacheFactory implements ItemStateCacheFactory {
+
+    /** The cache manager. */
+    private CacheManager cacheMgr;
+
+    /**
+     * Construct a new factory using a cache manager.
+     *
+     * @param cacheMgr the cache manager
+     */
+    public ManagedMLRUItemStateCacheFactory(CacheManager cacheMgr) {
+        this.cacheMgr = cacheMgr;
+    }
+
+    /**
+     * Create a new cache instance and link it to the cache manager.
+     */
+    public ItemStateCache newItemStateCache() {
+        MLRUItemStateCache cache = new MLRUItemStateCache();
+        cacheMgr.add(cache);
+        cache.setAccessListener(cacheMgr);
+        return cache;
+    }
+
+}
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/CacheAccessListener.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/CacheAccessListener.java	(revision 0)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/CacheAccessListener.java	(revision 0)
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.jackrabbit.core.state;
+
+/**
+ * The cache access listener can be registerd to a class.
+ * From time to time, the method cacheAccess is called.
+ */
+public interface CacheAccessListener {
+
+    /**
+     * The access listener is only called each x accesses.
+     */
+    static final int ACCESS_INTERVAL = 127;
+
+    /**
+     * The cache calls this method after a number of accessed.
+     */
+    void cacheAccessed();
+
+    /**
+     * Called after the cache is no longer used.
+     */
+    void disposeCache(Cache cache);
+
+}
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateCacheFactory.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateCacheFactory.java	(revision 0)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateCacheFactory.java	(revision 0)
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.jackrabbit.core.state;
+
+/**
+ * An item state cache factory to construct new item state caches.
+ */
+public interface ItemStateCacheFactory {
+
+    /**
+     * Construct a new item state cache.
+     *
+     * @return the new cache
+     */
+    ItemStateCache newItemStateCache();
+}
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/XAItemStateManager.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/XAItemStateManager.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/XAItemStateManager.java	(working copy)
@@ -86,8 +86,8 @@
      * @param factory        event state collection factory
      */
     public XAItemStateManager(SharedItemStateManager sharedStateMgr,
-                              EventStateCollectionFactory factory) {
-        this(sharedStateMgr, factory, DEFAULT_ATTRIBUTE_NAME);
+                              EventStateCollectionFactory factory, ItemStateCacheFactory cacheFactory) {
+        this(sharedStateMgr, factory, DEFAULT_ATTRIBUTE_NAME, cacheFactory);
     }
 
     /**
@@ -99,8 +99,9 @@
      */
     public XAItemStateManager(SharedItemStateManager sharedStateMgr,
                               EventStateCollectionFactory factory,
-                              String attributeName) {
-        super(sharedStateMgr, factory);
+                              String attributeName,
+                              ItemStateCacheFactory cacheFactory) {
+        super(sharedStateMgr, factory, cacheFactory);
 
         this.attributeName = attributeName;
     }
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateCache.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateCache.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/ItemStateCache.java	(working copy)
@@ -108,4 +108,9 @@
      * @param id the id of the item that was modified.
      */
     void update(ItemId id);
+
+    /**
+     * Informs the cache that it is no longer in use.
+     */
+    void dispose();
 }
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/CacheManager.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/CacheManager.java	(revision 0)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/state/CacheManager.java	(revision 0)
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+package org.apache.jackrabbit.core.state;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.WeakHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class manages the size of the caches used in Jackrabbit. The combined
+ * size of all caches must be limited to avoid out of memory problems. The
+ * available memory is dynamically distributed across the caches each second.
+ * This class tries to calculates the best cache sizes by comparing the access
+ * counts of each cache, and the used memory. The idea is, the more a cache is
+ * accessed, the more memory it should get, while the cache should not shrink
+ * too quickly. A minimum and maximum size per cache is defined as well. After
+ * distributing the memory in this way, there might be some unused memory (if
+ * one or more caches did not use some of the allocated memory). This unused
+ * memory is distributed evenly across the full caches.
+ *
+ */
+public class CacheManager implements CacheAccessListener {
+
+    /** The logger instance. */
+    private static Logger log = LoggerFactory.getLogger(CacheManager.class);
+
+    /** The amount of memory to distribute accross the caches. */
+    private static final long MAX_MEMORY = 16 * 1024 * 1024;
+
+    /** The minimum size of a cache. */
+    private static final long MIN_MEMORY_PER_CACHE = 128 * 1024;
+
+    /** The maximum memory per cache (unless, there is some unused memory). */
+    private static final long MAX_MEMORY_PER_CACHE = 4 * 1024 * 1024;
+
+    /** The set of caches (weakly referenced). */
+    private WeakHashMap caches = new WeakHashMap();
+
+    /** Rebalance the caches each ... milliseconds at most. */
+    private static final int SLEEP = 1000;
+
+    /** The size of a big object, to detect if a cache is full or not. */
+    private static final int BIG_OBJECT_SIZE = 16 * 1024;
+
+    /** The last time the caches where resized. */
+    private volatile long nextResize = System.currentTimeMillis() + SLEEP;
+
+    /**
+     * After one of the caches is accessed a number of times, this method is called.
+     * Resize the caches if required.
+     */
+    public void cacheAccessed() {
+        long now = System.currentTimeMillis();
+        if (now < nextResize) {
+            return;
+        }
+        synchronized (this) {
+            // the previous test was not synchronized (for speed)
+            // so we need another synchronized test
+            if (now < nextResize) {
+                return;
+            }
+            nextResize = now + SLEEP;
+            resizeAll();
+            nextResize = System.currentTimeMillis() + SLEEP;
+        }
+    }
+
+    /**
+     * Re-calcualte the maximum memory for each cache, and set the new limits.
+     */
+    private void resizeAll() {
+        log.info("resizeAll size="+ caches.size());
+        // get strong references
+        // entries in a weak hash map may disappear any time
+        // so can't use size() / keySet() directly
+        // only using the iterator guarantees that we don't get null references
+        ArrayList list = new ArrayList();
+        for (Iterator it = caches.keySet().iterator(); it.hasNext();) {
+            list.add(it.next());
+        }
+        if (list.size() == 0) {
+            // nothing to do
+            return;
+        }
+        CacheInfo[] infos = new CacheInfo[list.size()];
+        for (int i = 0; i < list.size(); i++) {
+            infos[i] = new CacheInfo((Cache) list.get(i));
+        }
+        // calculate the total access count and memory used
+        long totalAccessCount = 0;
+        long totalMemoryUsed = 0;
+        for (int i = 0; i < infos.length; i++) {
+            totalAccessCount += infos[i].getAccessCount();
+            totalMemoryUsed += infos[i].getMemoryUsed();
+        }
+        // try to distribute the memory based on the access count
+        // and memory used (higher numbers - more memory)
+        // and find out how many caches are full
+        // 50% is distributed according to access count,
+        // and 50% according to memory used
+        double memoryPerAccess = (double) MAX_MEMORY / 2.
+                / Math.max(1., (double) totalAccessCount);
+        double memoryPerUsed = (double) MAX_MEMORY / 2.
+                / Math.max(1., (double) totalMemoryUsed);
+        int fullCacheCount = 0;
+        for (int i = 0; i < infos.length; i++) {
+            CacheInfo info = infos[i];
+            long mem = (long) (memoryPerAccess * info.getAccessCount());
+            mem += (long) (memoryPerUsed * info.getMemoryUsed());
+            mem = Math.min(mem, MAX_MEMORY_PER_CACHE);
+            if (info.wasFull()) {
+                fullCacheCount++;
+            } else {
+                mem = Math.min(mem, info.getMemoryUsed());
+            }
+            mem = Math.min(mem, MAX_MEMORY_PER_CACHE);
+            mem = Math.max(mem, MIN_MEMORY_PER_CACHE);
+            info.setMemory(mem);
+        }
+        // calculate the unused memory
+        long unusedMemory = MAX_MEMORY;
+        for (int i = 0; i < infos.length; i++) {
+            unusedMemory -= infos[i].getMemory();
+        }
+        // distribute the remaining memory evenly across the full caches
+        if (unusedMemory > 0 && fullCacheCount > 0) {
+            for (int i = 0; i < infos.length; i++) {
+                CacheInfo info = infos[i];
+                if (info.wasFull()) {
+                    info.setMemory(info.getMemory() + unusedMemory
+                            / fullCacheCount);
+                }
+            }
+        }
+        // set the new limit
+        for (int i = 0; i < infos.length; i++) {
+            CacheInfo info = infos[i];
+            Cache cache = info.getCache();
+            log.debug(cache + " now:" + cache.getMaxMemorySize() + " used:"
+                    + info.getMemoryUsed() + " access:" + info.getAccessCount()
+                    + " new:" + info.getMemory());
+            cache.setMaxMemorySize(info.getMemory());
+        }
+    }
+
+    /**
+     * Add a new cache to the list.
+     * This call does not trigger recalculating the cache sizes.
+     *
+     * @param cache the cache to add
+     */
+    public synchronized void add(Cache cache) {
+        caches.put(cache, null);
+    }
+
+    /**
+     * Remove a cache. As this class only has a weak reference to each cache,
+     * calling this method is not strictly required.
+     * This call does not trigger recalculating the cache sizes.
+     *
+     * @param cache
+     *            the cache to remove
+     */
+    public synchronized void remove(Cache cache) {
+        caches.remove(cache);
+    }
+
+    /**
+     * Internal copy of the cache information.
+     */
+    public static class CacheInfo {
+        private Cache cache;
+
+        private long accessCount;
+
+        private long memory;
+
+        private long memoryUsed;
+
+        private boolean wasFull;
+
+        CacheInfo(Cache cache) {
+            this.cache = cache;
+            // copy the data as this runs in a different thread
+            // the exact values are not important, but it is important that the
+            // values don't change
+            this.memory = cache.getMaxMemorySize();
+            this.memoryUsed = cache.getMemoryUsed();
+            this.accessCount = cache.getAccessCount();
+            // reset the access count, so that concurrent cache access is not lost
+            cache.resetAccessCount();
+            // if the memory used plus one large object is smaller than the
+            // allocated memory,
+            // then the memory was not fully used
+            wasFull = (memoryUsed + BIG_OBJECT_SIZE) >= memory;
+        }
+
+        boolean wasFull() {
+            return wasFull;
+        }
+
+        long getAccessCount() {
+            return accessCount;
+        }
+
+        long getMemoryUsed() {
+            return memoryUsed;
+        }
+
+        void setMemory(long mem) {
+            this.memory = mem;
+        }
+
+        long getMemory() {
+            return memory;
+        }
+
+        Cache getCache() {
+            return cache;
+        }
+
+    }
+
+    public void disposeCache(Cache cache) {
+        remove(cache);
+    }
+
+}
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/RepositoryImpl.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/RepositoryImpl.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/RepositoryImpl.java	(working copy)
@@ -45,6 +45,9 @@
 import org.apache.jackrabbit.core.state.ItemStateException;
 import org.apache.jackrabbit.core.persistence.PMContext;
 import org.apache.jackrabbit.core.persistence.PersistenceManager;
+import org.apache.jackrabbit.core.state.CacheManager;
+import org.apache.jackrabbit.core.state.ItemStateCacheFactory;
+import org.apache.jackrabbit.core.state.ManagedMLRUItemStateCacheFactory;
 import org.apache.jackrabbit.core.state.SharedItemStateManager;
 import org.apache.jackrabbit.core.state.ChangeLog;
 import org.apache.jackrabbit.core.version.VersionManager;
@@ -210,6 +213,16 @@
     private final ReadWriteLock shutdownLock = new WriterPreferenceReadWriteLock();
 
     /**
+     * There is one cache manager per repository that manages the sizes of the caches used.
+     */
+    private final CacheManager cacheMgr = new CacheManager();
+
+    /**
+     * There is only one item state cache factory
+     */
+    private final ItemStateCacheFactory cacheFactory = new ManagedMLRUItemStateCacheFactory(cacheMgr);
+
+    /**
      * private constructor
      *
      * @param repConfig
@@ -321,6 +334,15 @@
     }
 
     /**
+     * Get the item state cache factory of this repository.
+     *
+     * @return the cache manager
+     */
+    public ItemStateCacheFactory getItemStateCacheFactory() {
+        return cacheFactory;
+    }
+
+    /**
      * Creates the version manager.
      *
      * @param vConfig the versioning config
@@ -341,7 +363,7 @@
                 ntReg);
 
         VersionManagerImpl vMgr = new VersionManagerImpl(pm, fs, ntReg, delegatingDispatcher,
-                VERSION_STORAGE_NODE_ID, SYSTEM_ROOT_NODE_ID);
+                VERSION_STORAGE_NODE_ID, SYSTEM_ROOT_NODE_ID, cacheFactory);
         if (clusterNode != null) {
             vMgr.setEventChannel(clusterNode.createUpdateChannel());
         }
@@ -1703,7 +1725,7 @@
                 // create item state manager
                 try {
                     itemStateMgr =
-                            new SharedItemStateManager(persistMgr, rootNodeId, ntReg, true);
+                            new SharedItemStateManager(persistMgr, rootNodeId, ntReg, true, cacheFactory);
                     try {
                         itemStateMgr.addVirtualItemStateProvider(
                                 vMgr.getVirtualItemStateProvider());
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/version/VersionManagerImpl.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/version/VersionManagerImpl.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/version/VersionManagerImpl.java	(working copy)
@@ -21,24 +21,25 @@
 import org.apache.jackrabbit.core.NodeImpl;
 import org.apache.jackrabbit.core.PropertyId;
 import org.apache.jackrabbit.core.SessionImpl;
-import org.apache.jackrabbit.core.persistence.PersistenceManager;
+import org.apache.jackrabbit.core.cluster.UpdateEventChannel;
+import org.apache.jackrabbit.core.cluster.UpdateEventListener;
 import org.apache.jackrabbit.core.fs.FileSystem;
 import org.apache.jackrabbit.core.nodetype.NodeTypeRegistry;
 import org.apache.jackrabbit.core.observation.DelegatingObservationDispatcher;
 import org.apache.jackrabbit.core.observation.EventStateCollection;
 import org.apache.jackrabbit.core.observation.EventStateCollectionFactory;
+import org.apache.jackrabbit.core.persistence.PersistenceManager;
 import org.apache.jackrabbit.core.state.ChangeLog;
+import org.apache.jackrabbit.core.state.ItemState;
+import org.apache.jackrabbit.core.state.ItemStateCacheFactory;
 import org.apache.jackrabbit.core.state.ItemStateException;
+import org.apache.jackrabbit.core.state.ItemStateListener;
 import org.apache.jackrabbit.core.state.LocalItemStateManager;
 import org.apache.jackrabbit.core.state.NodeReferences;
 import org.apache.jackrabbit.core.state.NodeReferencesId;
 import org.apache.jackrabbit.core.state.NodeState;
 import org.apache.jackrabbit.core.state.PropertyState;
 import org.apache.jackrabbit.core.state.SharedItemStateManager;
-import org.apache.jackrabbit.core.state.ItemStateListener;
-import org.apache.jackrabbit.core.state.ItemState;
-import org.apache.jackrabbit.core.cluster.UpdateEventChannel;
-import org.apache.jackrabbit.core.cluster.UpdateEventListener;
 import org.apache.jackrabbit.core.value.InternalValue;
 import org.apache.jackrabbit.core.virtual.VirtualItemStateProvider;
 import org.apache.jackrabbit.name.MalformedPathException;
@@ -129,7 +130,8 @@
     public VersionManagerImpl(PersistenceManager pMgr, FileSystem fs,
                               NodeTypeRegistry ntReg,
                               DelegatingObservationDispatcher obsMgr, NodeId rootId,
-                              NodeId rootParentId) throws RepositoryException {
+                              NodeId rootParentId,
+                              ItemStateCacheFactory cacheFactory) throws RepositoryException {
         try {
             this.pMgr = pMgr;
             this.fs = fs;
@@ -156,9 +158,9 @@
                 pMgr.store(cl);
             }
             sharedStateMgr =
-                    new VersionItemStateManager(pMgr, rootId, ntReg);
+                    new VersionItemStateManager(pMgr, rootId, ntReg, cacheFactory);
 
-            stateMgr = new LocalItemStateManager(sharedStateMgr, escFactory);
+            stateMgr = new LocalItemStateManager(sharedStateMgr, escFactory, cacheFactory);
             stateMgr.addListener(this);
 
             NodeState nodeState = (NodeState) stateMgr.getItemState(rootId);
@@ -515,9 +517,10 @@
 
         public VersionItemStateManager(PersistenceManager persistMgr,
                                        NodeId rootNodeId,
-                                       NodeTypeRegistry ntReg)
+                                       NodeTypeRegistry ntReg,
+                                       ItemStateCacheFactory cacheFactory)
                 throws ItemStateException {
-            super(persistMgr, rootNodeId, ntReg, false);
+            super(persistMgr, rootNodeId, ntReg, false, cacheFactory);
         }
 
         protected void checkReferentialIntegrity(ChangeLog changes)
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/version/XAVersionManager.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/version/XAVersionManager.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/version/XAVersionManager.java	(working copy)
@@ -24,17 +24,18 @@
 import org.apache.jackrabbit.core.TransactionContext;
 import org.apache.jackrabbit.core.TransactionException;
 import org.apache.jackrabbit.core.nodetype.NodeTypeRegistry;
+import org.apache.jackrabbit.core.observation.EventStateCollection;
 import org.apache.jackrabbit.core.observation.EventStateCollectionFactory;
-import org.apache.jackrabbit.core.observation.EventStateCollection;
 import org.apache.jackrabbit.core.state.ChangeLog;
 import org.apache.jackrabbit.core.state.ItemState;
+import org.apache.jackrabbit.core.state.ItemStateCacheFactory;
 import org.apache.jackrabbit.core.state.ItemStateException;
+import org.apache.jackrabbit.core.state.ItemStateListener;
 import org.apache.jackrabbit.core.state.NoSuchItemStateException;
 import org.apache.jackrabbit.core.state.NodeReferences;
 import org.apache.jackrabbit.core.state.NodeReferencesId;
 import org.apache.jackrabbit.core.state.NodeState;
 import org.apache.jackrabbit.core.state.XAItemStateManager;
-import org.apache.jackrabbit.core.state.ItemStateListener;
 import org.apache.jackrabbit.core.virtual.VirtualItemStateProvider;
 import org.apache.jackrabbit.core.virtual.VirtualNodeState;
 import org.apache.jackrabbit.core.virtual.VirtualPropertyState;
@@ -96,14 +97,14 @@
      * Creates a new instance of this class.
      */
     public XAVersionManager(VersionManagerImpl vMgr, NodeTypeRegistry ntReg,
-                            SessionImpl session)
+                            SessionImpl session, ItemStateCacheFactory cacheFactory)
             throws RepositoryException {
 
         this.vMgr = vMgr;
         this.ntReg = ntReg;
         this.session = session;
         this.stateMgr = new XAItemStateManager(vMgr.getSharedStateMgr(),
-                this, CHANGE_LOG_ATTRIBUTE_NAME);
+                this, CHANGE_LOG_ATTRIBUTE_NAME, cacheFactory);
 
         NodeState state;
         try {
@@ -199,6 +200,7 @@
      * {@inheritDoc}
      */
     public void close() throws Exception {
+        stateMgr.dispose();
     }
 
     //---------------------------------------------< VirtualItemStateProvider >
Index: C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/WorkspaceImpl.java
===================================================================
--- C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/WorkspaceImpl.java	(revision 472064)
+++ C:/dev/jackrabbit/src/main/java/org/apache/jackrabbit/core/WorkspaceImpl.java	(working copy)
@@ -21,13 +21,12 @@
 import org.apache.jackrabbit.core.lock.LockManager;
 import org.apache.jackrabbit.core.observation.EventStateCollection;
 import org.apache.jackrabbit.core.observation.EventStateCollectionFactory;
-import org.apache.jackrabbit.core.observation.ObservationDispatcher;
 import org.apache.jackrabbit.core.observation.ObservationManagerImpl;
 import org.apache.jackrabbit.core.query.QueryManagerImpl;
 import org.apache.jackrabbit.core.state.LocalItemStateManager;
 import org.apache.jackrabbit.core.state.SharedItemStateManager;
+import org.apache.jackrabbit.core.version.DateVersionSelector;
 import org.apache.jackrabbit.core.version.VersionImpl;
-import org.apache.jackrabbit.core.version.DateVersionSelector;
 import org.apache.jackrabbit.core.version.VersionSelector;
 import org.apache.jackrabbit.core.xml.ImportHandler;
 import org.apache.jackrabbit.core.xml.Importer;
@@ -168,6 +167,7 @@
             obsMgr.dispose();
             obsMgr = null;
         }
+        stateMgr.dispose();
     }
 
     /**
@@ -757,7 +757,7 @@
      * @return local item state manager
      */
     protected LocalItemStateManager createItemStateManager(SharedItemStateManager shared) {
-        return new LocalItemStateManager(shared, this);
+        return new LocalItemStateManager(shared, this, rep.getItemStateCacheFactory());
     }
 
     //------------------------------------------< EventStateCollectionFactory >