/*
 * $Header: /home/cvspublic/jakarta-slide/src/share/org/apache/slide/util/ByteSizeLimitedObjectCache.java,v 1.8 2005/06/02 17:25:37 ozeigermann Exp $
 * $Revision: 1.8 $
 * $Date: 2005/06/02 17:25:37 $
 *
 * ====================================================================
 *
 * Copyright 1999-2002 The Apache Software Foundation 
 *
 * 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.
 *
 */

package org.apache.slide.util;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.commons.collections.LRUMap;
import org.apache.slide.util.logger.Logger;

/**
 * Transactional LRU byte counting object cache. Caches objects using a least-recently-used strategy. 
 * Asserts a given overall size of all cached objects does not exceed a specified limit.   
 * 
 * It provides basic isolation from other transactions and atomicity of all operations. As
 * no locking is used (which is undesirable mainly as a cache should never block and a commit must never fail) 
 * serializability needs to be guaranteed by underlying stores. 
 * 
 * <em>Note</em>: Unlike {@link TxLRUObjectCache} this cache also limits the size of temporary caches inside transactions.
 * This is necessary as size of bytes in a transaction can become really big. In case an entry can not be cached inisde a 
 * transaction it will be marked as invalid after commit of that tx.
 *  
 * @version $Revision: 1.8 $
 * @see TxLRUObjectCache
 */
public class ByteSizeLimitedObjectCache extends TxLRUObjectCache {

    // amount of entries that are at most removed from cache to make room for a new entry
    protected static final int MAX_FREEING_TRIES = 10;

    protected long globalByteSize;
    protected int txCacheSize;
    protected long txByteSize;
    protected long maxByteSizePerEntry;

    /**
     * Creates a new object cache. The maximum cache size can be configured for local transaction
     * caches as well as on a global level. 
     * 
     * The idea of having a maximum size in bytes per entry is to prevent a large entry
     * to displace many small entries.
     * 
     * @param globalCacheSize maximum size in objects of global cache
     * @param txCacheSize maximum size in objects for local transaction cache 
     * @param globalByteSize maximum size in bytes of global cache
     * @param txByteSize maximum size in bytes for local transaction cache
     * @param maxByteSizePerEntry maximum size of a single cache entry in bytes
     * @param name the name used to construct logging category / channel
     * @param logger Slide logger to be used for logging 
     * @param noGlobalCachingInsideTx indicates global caches are enabled, but shall not be used inside transactions
     */
    public ByteSizeLimitedObjectCache(
        int globalCacheSize,
        int txCacheSize,
        long globalByteSize,
        long txByteSize,
        long maxByteSizePerEntry,
        String name,
        Logger logger,
        boolean noGlobalCachingInsideTx) {
        super(globalCacheSize, name, logger, noGlobalCachingInsideTx);
        globalCache = new SizeCountingLRUMap(globalCacheSize, globalByteSize, maxByteSizePerEntry);
        this.globalByteSize = globalByteSize;
        this.txCacheSize = txCacheSize;
        this.txByteSize = txByteSize;
        this.maxByteSizePerEntry = maxByteSizePerEntry;
        logChannel = "ByteSizeLimitedObjectCache";
        if (name != null) {
            logChannel += "." + name;
        }

    }

    public synchronized void clear() {
        super.clear();
    }

    public synchronized boolean canCache(Object txId, long byteSize) {
        long maxSize;
        if (txId == null) {
            maxSize = globalByteSize;
        } else {
            maxSize = txByteSize;
        }
        return (maxSize >= byteSize && maxByteSizePerEntry >= byteSize);
    }

    public synchronized void put(Object txId, Object key, Object value, long byteSize, long timeout) {
        if (timeout != -1) {
            value = new TimeoutValuePair(value, timeout);
        }
        put(txId, key, value, byteSize);
    }
    
    public synchronized void put(Object txId, Object key, Object value, long byteSize) {
        if (key == null) {
            logger.log(txId + " adding null key with byte size " + byteSize, logChannel, Logger.WARNING);
        }

        if (txId != null) {
            // if it has been deleted before, undo this
            Set deleteCache = (Set) txDeleteCaches.get(txId);
            deleteCache.remove(key);

            SizeCountingLRUMap changeCache = (SizeCountingLRUMap) txChangeCaches.get(txId);
            changeCache.put(key, value, byteSize);

            if (loggingEnabled) {
                logger.log(txId + " added '" + key + "' with byte size " + byteSize, logChannel, Logger.DEBUG);
            }
        } else {
            ((SizeCountingLRUMap) globalCache).put(key, value, byteSize);
            if (loggingEnabled) {
                logger.log("Added '" + key + "' with byte size " + byteSize, logChannel, Logger.DEBUG);
            }
        }
    }

    public synchronized void start(Object txId) {
        if (txId != null) {
            txChangeCaches.put(txId, new SizeCountingLRUMap(txCacheSize, txByteSize, maxByteSizePerEntry, txId));
            txDeleteCaches.put(txId, new HashSet());
        }
    }

    public synchronized void commit(Object txId) {
        if (txId != null) {
            // apply local changes and deletes (is atomic as we have a global lock on this TxCache)

            SizeCountingLRUMap changeCache = (SizeCountingLRUMap) txChangeCaches.get(txId);
            for (Iterator it = changeCache.entrySet().iterator(); it.hasNext();) {
                Map.Entry entry = (Map.Entry) it.next();
                long byteSize = changeCache.getByteSize(entry.getKey());
                if (byteSize == -1L) {
                    putToGlobalCache(entry.getKey(), entry.getValue());
                } else {
                    ((SizeCountingLRUMap) globalCache).put(entry.getKey(), entry.getValue(), byteSize);
                }
            }

            Set deleteCache = (Set) txDeleteCaches.get(txId);
            for (Iterator it = deleteCache.iterator(); it.hasNext();) {
                Object key = it.next();
                globalCache.remove(key);
            }

            if (loggingEnabled) {
                logger.log(
                    txId
                        + " committed "
                        + changeCache.size()
                        + " changes and "
                        + deleteCache.size()
                        + " scheduled deletes",
                    logChannel,
                    Logger.DEBUG);
            }

            // finally forget about tx
            forget(txId);
        }
    }

    protected class SizeCountingLRUMap extends LRUMap {

        private Map shadowSizes;
        private long globalSize;
        private long maxByteSize;
        private long maxByteSizePerEntry;
        private Object txId;

        public SizeCountingLRUMap(int size, long maxByteSize, long maxByteSizePerEntry, Object txId) {
            this(size, maxByteSize, maxByteSizePerEntry);
            this.txId = txId;
        }

        public SizeCountingLRUMap(int size, long maxByteSize, long maxByteSizePerEntry) {
            super(size);
            this.shadowSizes = new HashMap();
            this.globalSize = 0;
            this.maxByteSize = maxByteSize;
            this.maxByteSizePerEntry = maxByteSizePerEntry;
        }

        public Object put(Object key, Object value, long byteSize) {
            // is it too big to be cached?
            if (byteSize > maxByteSizePerEntry || byteSize > maxByteSize) {
                if (loggingEnabled) {
                    logger.log(txId + " for '" + key + "' is too big to be cached", logChannel, Logger.DEBUG);
                }
                Object oldValue = get(key);
                // invalidate previous entry if present
                // XXX this relies on an implementation detail in TxLRUByteCache.put
                // removal from delete cache must be done before trying to add to
                // change cache using this method; if not our undoing will be partly
                // undone (again) in TxLRUObjectCache.put
                invalidate(key);
                return oldValue;
            } else {
                // be sure to return allocated bytes before readding
                freeBytes(key);

                // ok, we decided to cache this entry, make room for it
                for (int i = 0; globalSize + byteSize > maxByteSize && i < MAX_FREEING_TRIES; i++) {
                    if (loggingEnabled) {
                        logger.log(txId + " for '" + key + "' needs "
                                + Long.toString(globalSize + byteSize - maxByteSize)
                                + " bytes more to be cached. Freeing bytes!", logChannel, Logger.DEBUG);
                    }
                    // this will call back processRemovedLRU and will thus free
                    // bytes
                    removeLRU();
                }
            }
            // was this successful?
            if (globalSize + byteSize <= maxByteSize) {
                shadowSizes.put(key, new Long(byteSize));
                globalSize += byteSize;
                return super.put(key, value);
            } else {
                Object oldValue = get(key);
                invalidate(key);
                return oldValue;
            }
        }

        public long getByteSize(Object key) {
            Long lSize = (Long) shadowSizes.get(key);
            if (lSize != null) {
                return lSize.longValue();
            } else {
                return -1L;
            }
        }

        protected void freeBytes(Object key) {
            Long lSize = (Long) shadowSizes.remove(key);
            if (lSize != null) {
                long size = lSize.longValue();
                globalSize -= size;
            }
        }

        protected void invalidate(Object key) {
            // if it is no longer cached, it still means entry currently
            // in global map gets invalid upon commit of this tx
            freeBytes(key);
            ByteSizeLimitedObjectCache.this.remove(txId, key);
            if (loggingEnabled) {
                logger.log(txId + " invalidated '" + key + "'", logChannel, Logger.DEBUG);
            }
        }

        // notify cache that this entry has been removed by LRU strategy (called back by LRUMap)
        protected void processRemovedLRU(Object key, Object value) {
            invalidate(key);
        }
    }
}
