Quadcap Embeddable Database

com/quadcap/sql/file/Logger1.java

Go to the documentation of this file.

package com.quadcap.sql.file;

/* Copyright 1999 - 2003 Quadcap Software.  All rights reserved.
 *
 * This software is distributed under the Quadcap Free Software License.
 * This software may be used or modified for any purpose, personal or
 * commercial.  Open Source redistributions are permitted.  Commercial
 * redistribution of larger works derived from, or works which bundle
 * this software requires a "Commercial Redistribution License"; see
 * http://www.quadcap.com/purchase.
 *
 * Redistributions qualify as "Open Source" under  one of the following terms:
 *   
 *    Redistributions are made at no charge beyond the reasonable cost of
 *    materials and delivery.
 *
 *    Redistributions are accompanied by a copy of the Source Code or by an
 *    irrevocable offer to provide a copy of the Source Code for up to three
 *    years at the cost of materials and delivery.  Such redistributions
 *    must allow further use, modification, and redistribution of the Source
 *    Code under substantially the same terms as this license.
 *
 * Redistributions of source code must retain the copyright notices as they
 * appear in each source code file, these license terms, and the
 * disclaimer/limitation of liability set forth as paragraph 6 below.
 *
 * Redistributions in binary form must reproduce this Copyright Notice,
 * these license terms, and the disclaimer/limitation of liability set
 * forth as paragraph 6 below, in the documentation and/or other materials
 * provided with the distribution.
 *
 * The Software is provided on an "AS IS" basis.  No warranty is
 * provided that the Software is free of defects, or fit for a
 * particular purpose.  
 *
 * Limitation of Liability. Quadcap Software shall not be liable
 * for any damages suffered by the Licensee or any third party resulting
 * from use of the Software.
 */

import java.io.BufferedInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.RandomAccessFile;

import java.util.Properties;

import com.quadcap.sql.io.ObjectOutputStream;
import com.quadcap.sql.io.ObjectInputStream;

import com.quadcap.util.collections.LongMap;

import com.quadcap.util.Debug;
import com.quadcap.util.Util;

/**
 * A Logger implmeentation using a fixed size circular buffer.
 *
 * @author Stan Bailes
 */
public class Logger1 implements Logger {
    LogBuffer                           cb;
    InputStream                         cbIn = null;
    ObjectOutputStream                  oos = null;
    ByteArrayRandomAccess               bra = new ByteArrayRandomAccess();
    RandomAccessInputStream             ris = new RandomAccessInputStream(bra);
    ObjectInputStream                   ois = new ObjectInputStream(ris);
    byte[]                              tmp = new byte[8];

    /**
     * Transaction map:
     *   - ordered by transactionStart
     *   - first transaction is the oldest transaction still running
     *        so when committing the oldest transaction, move the begin
     *   - last transaction is the newest transaction still running
     * transactions!
     */
    TransMap[]                          trans = new TransMap[16];

    /**
     * Keeps track of the number of TransMap entries allocated.  Some entries
     * will correspond to completed but not yet checkpointed transactions,
     * so this number may be larger than the number of actual live transactions.
     */
    int                                 numTrans;

    /**
     * Keep the actual count of live transactions as determined by calls
     * to beginTransaction, endTransaction.
     */
    int                                 liveTrans       = 0;

    /**
     * The starting position of the most recent op we wrote to the file
     */
    long                                prevOp;

    int                                 maxSize = 128 * 1024 * 1024;
    int                                 minSize = 128 * 1024;

    Log                                 myLog;
    Datafile                            db;

    Logger1() {}

    public void init(Log log, boolean create, Properties props)
        throws IOException
    {
        this.myLog = log;
        this.db = log.getDatafile();
        this.cb = new LogBuffer();
        // XXX better would be a call to Datafile.getProperty("logfile")
        /*{com.quadcap.sql.Datafile-conn.xml-29}
         * <config-var>
         *   <config-name>maxLogSize</config-name>
         *   <config-dflt>128 M bytes</config-dflt>
         *   <config-desc>For loggers which support rollback, the maximum
         *     size of the rollback log.  Once this maximum is reached, it
         *     may be necessary to abort the oldest transaction and checkpoint
         *     in an attempt to regain log space.</config-desc>
         * </config-var>
         */
        maxSize = Integer.parseInt(props.getProperty("maxLogSize", "" + maxSize));

        /*{com.quadcap.sql.Datafile-conn.xml-29}
         * <config-var>
         *   <config-name>minLogSize</config-name>
         *   <config-dflt>128 K bytes</config-dflt>
         *   <config-desc>For loggers which support rollback, the minimum size
         *     at which a logging checkpoint operation will occur.  The logging
         *     checkpoint cleans the log of entries which correspond to completed
         *     transactions.</config-desc>
         * </config-var>
         */
        minSize = Integer.parseInt(props.getProperty("minLogSize", "" + minSize));

        File logfile = new File(db.getDbRootDir(), "logfile");
        RandomAccessFile raf = new RandomAccessFile(logfile, "rw");
        FileRandomAccess fra = new FileRandomAccess(raf, maxSize);
        RandomAccess ra = fra;
        if (create) {
            cb.init(ra, maxSize);
        } else {
            lastSize = ra.size();
            cb.init(ra, props);
        }
    }

    public void init(File file) throws IOException {
        RandomAccessFile raf = new RandomAccessFile(file, "rw");
        FileRandomAccess fra = new FileRandomAccess(raf, raf.length());
        RandomAccess ra = fra;
        this.cb = new LogBuffer();
        lastSize = ra.size();
        cb.init(ra, new Properties());
     }


    /**
     * Inner class used to to track active transactions.
     */
    class TransMap {
        long                    transId;
        /**
         * If >0, the position of the first op in the transaction.
         * This will generally be the BEGIN_TRANSACTION op.
         */
        int                     bufStart;
        /**
         * If >0, the position of the most recently written op in the
         * transaction.  
         */
        int                     bufEnd;
        boolean                 complete;
        
        public void init(long t, int p) {
            this.transId = t;
            this.bufStart = p;
            this.bufEnd = p;
            this.complete = false;
        }

        public void init(TransMap t) {
            this.transId = t.transId;
            this.bufStart = -1;
            this.bufEnd = -1;
            this.complete = t.complete;
        }

        public void copy(TransMap t) {
            this.transId = t.transId;
            this.bufStart = t.bufStart;
            this.bufEnd = t.bufEnd;
            this.complete = t.complete;
        }

        public String toString() {
            return "TransMap[" + transId + ": " + bufStart + "-" + bufEnd +
                (complete ? " (COMPLETE)" : "") + "]";
        }
    }

    public String toString() {
        StringBuffer sb = new StringBuffer("Logger1 {" + cb.getCheckpoint() +
                                           ", " + cb.getEnd() + "} ");
        for (int i = 0; i < trans.length; i++) {
            TransMap tm = trans[i];
            if (tm != null && !tm.complete) {
                sb.append(tm.toString());
            }
        }
        return sb.toString();
    }

    public int getCheckpoint() {
        return cb.getCheckpoint();
    }

    public int getEnd() {
        return cb.getEnd();
    }
    
    public int rput(LogEntry op) throws IOException {
        int pos = cb.getEnd();
        op.setPosition(pos);

        long id = op.getTransactionId();
        int tx = findTransaction(id);
        if (tx >= 0) {
            TransMap tm = trans[tx];
            op.setPrev(tm.bufEnd);
            tm.bufEnd = pos;
        }

        try {
            if (oos == null) {
                oos = new ObjectOutputStream(cb.getOutputStream());
            }
            oos.writeObject(op);
            oos.flush();
        } catch (IOException ex) {
            if (tx >= 0) {
                trans[tx].bufEnd = op.getPrev();
            }
            throw ex;
        }
        return pos;
    }
    
    public void put(LogEntry op) throws IOException {
        int pos = rput(op);
        switch (op.getCode()) {
        case LogEntry.BEGIN_TRANSACTION:
            beginTransaction(op.getTransactionId(), pos);
            break;
        case LogEntry.COMMIT:
            endTransaction(op.getTransactionId());
            break;
        }
        //Debug.println("Logger1.put(" + op + ") [" + pos +
        //              "] ----------------------");
    }
    
    public void setRedoState(LogEntry op, int state) throws IOException {
        int pos = op.getPosition();
        //Debug.println("setRedoState(" + op + ", " + state + ")"); 
        cb.writeByte(pos+1, (byte)state);
        op.redoState = state;
    }
    
    public LogEntry getLastOp(long transId) throws IOException {
        LogEntry ret = null;
        int tx = findTransaction(transId);
        if (tx >= 0) {
            TransMap tm = trans[tx];
            ret = readEntry(tm.bufEnd);
        }
        //#ifdef DEBUG
        if (ret == null) {
            Debug.println(Util.stackTrace());
            Debug.println("getLastOp(" + transId + ") = null, trans:");
            for (int i = 0; i < numTrans; i++) {
                Debug.println("trans[" + i + "] = " + trans[i]);
            }
        }
        //#endif
        return ret;
    }

    public LogEntry getPrevOp(LogEntry op) throws IOException {
        int pos = op.getPrev();
        int len = op.getPosition() - pos;
        LogEntry ret = readEntry(pos, len);
        return ret;        
    }

    public LogEntry getFirstOp() throws IOException {
        LogEntry ret = readEntry(cb.getBegin());
        return ret;
    }

    public LogEntry getNextOp() throws IOException {
        return readEntry();
    }

    public void sync() throws IOException {
        cb.sync();
    }

    // close all the files
    public void close() throws IOException {
        cb.close();
    }

    long lastSize = 0;
    public void reset() throws IOException {        
        cb.reset();
        if (lastSize - cb.size() > 100 * 1000) {
            cb.truncate();
        }
        lastSize = cb.size();
    }

    public long getOldestTransaction() {
        if (numTrans > 0) {
            return trans[0].transId;
        } else {
            return -1;
        }
    }

    public int getActiveTransactionCount() {
        return liveTrans;
    }

    public LongMap getActiveTransactions() {
        LongMap map = new LongMap(16);
        for (int i = 0; i < numTrans; i++) {
            TransMap t = trans[i];
            if (!t.complete) {
                map.put(t.transId, "");
            }
        }
        return map;
    }

    public void checkpoint() throws IOException {
        int offset = 0;
        for (int i = 0; i < numTrans; i++) {
            TransMap t = trans[i];
            if (t.complete) {
                offset++;
            } else if (offset > 0) {
                trans[i-offset].init(t);
            }
        }
        //#ifdef DEBUG
        if (Trace.bit(22)) {
            Debug.println("BEGIN checkpoint [" + cb.getBegin() + "-" +
                          cb.getEnd() + " (" + numTrans + " transactions)]");
        }
        //#endif
        numTrans -= offset;
        //Debug.println("Checkpoint: -" + offset + " = " + numTrans);
        LogEntry entry = readEntry(0);
        cb.reset();
        if (numTrans > 0) {
            while (entry != null) {
                boolean keep = false;
                long id = entry.getTransactionId();
                int tx = findTransaction(id);
                if (tx >= 0) {
                    TransMap tm = trans[tx];
                    keep = !tm.complete;
                    if (keep) {
                        if (tm.bufStart < 0) tm.bufStart = cb.getEnd();
                    }
                }
                if (keep) {
                    //#ifdef DEBUG
                    if (Trace.bit(23)) {
                        Debug.println("   [" + cb.getEnd() + "] -> " + entry);
                    }
                    //#endif
                    rput(entry);
                } else {
                    //#ifdef DEBUG
                    if (Trace.bit(23)) {
                        Debug.println("   [" + cb.getEnd() + "] -> DISCARD: " + entry);
                    }
                    //#endif
                    entry.discard(db);
                }
                entry = readEntry();
            }
        }
        cb.checkpoint();
        //#ifdef DEBUG
        if (Trace.bit(22)) {
            Debug.println("END checkpoint [" + cb.getBegin() + "-" + cb.getEnd() +
                          " (" + numTrans + " transactions)]");
        }
        //#endif
    }

    //--------------------------------- private stuff
    
    private final void beginTransaction(long transId, int pos) {
        if (numTrans >= trans.length) {
            int newcap  = numTrans + (numTrans/2) + 2;
            trans = (TransMap[])Util.checkCapacity(trans, newcap);
        }
        TransMap t = trans[numTrans];
        if (t == null) {
            t = new TransMap();
            trans[numTrans] = t;
        }

        // A race condition exists -- this might *not* be the newest trans,
        // and we want to keep the 'trans' array ordered properly, so search
        // backwards from the end to make sure we put this fellow in the
        // right place.
        for (int i = numTrans-1; i >= 0; i--) {
            if (trans[i].transId > transId) {
                trans[i+1].copy(trans[i]);
                t = trans[i];
            } else {
                break;
            }
        }
        numTrans++;
        //Debug.println("Begin T:" + transId + " = " + numTrans);
        liveTrans++;
        t.init(transId, pos);
        //Debug.println("beginTransaction(" + transId + "): " + numTrans);
    }

    private final void endTransaction(final long transId) throws IOException {
        //Debug.println("endTransaction(" + transId + "): " + numTrans);
        int pos = findTransaction(transId);
        if (pos >= 0) {
            // Mark the transaction as complete.  We don't decrement
            // numTrans here; we wait until the next checkpoint, where
            // all of the completed transactions can be reclaimed at
            // once.
            trans[pos].complete = true;
            liveTrans--;
            if (pos == 0 && cb.size() > minSize) {
                checkpoint();
            }
        }
    }

    /**
     * Find the TransMap entry for the given transaction id
     */
    int findTransaction(long transId) {
        int lo = 0;
        int hi = numTrans-1;
        while (hi >= lo) {
            int mid = (hi+lo) / 2;
            TransMap t = trans[mid];
            if (t.transId == transId) {
                return mid;
            }
            if (t.transId < transId) {
                lo = mid+1;
            } else {
                hi = mid-1;
            }
        }
        return -1;
    }

    private final LogEntry readEntry(int pos, int len) throws IOException {
        if (pos < 0) return null;
        bra.resize(len);
        byte[] buf = bra.getBytes();
        cb.read(pos, buf, 0, len);
        ris.setPosition(0);
        ois.setInputStream(ris);
        ois.setPosition(pos);
        
        LogEntry ret = readEntry();
        return ret;
    }

    private final LogEntry readEntry(int pos) throws IOException {
        if (pos < 0) return null;
        cbIn = new BufferedInputStream(cb.getInputStream(pos));
        ois.setInputStream(cbIn);
        ois.setPosition(pos);

        LogEntry ret = readEntry();
        return ret;
    }

    private final LogEntry readEntry() throws IOException {
        Object obj = null;
        try {
            int pos = (int)ois.getPosition();
            LogEntry ret = (LogEntry)(obj = ois.readObject());
            if (ret != null) {
                ret.setPosition(pos);
            }
            return ret;
        } catch (EOFException ex) {
            return null;
        } catch (ClassCastException ez) {
            throw new DatafileException(ez);
        } catch (ClassNotFoundException ex) {
            throw new DatafileException(ex);
        }
    }
}