Quadcap Embeddable Database

com/quadcap/sql/file/SubPageManager.java

Go to the documentation of this file.

package com.quadcap.sql.file;

/* Copyright 1997 - 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.IOException;
import java.io.InputStream;
import java.io.OutputStream;

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

/**
 * This class implements a sub-page manager which operates by allocating
 * pages as needed from the main file and chopping them up into smaller
 * pages of the appropriate size. 
 *
 * @author Stan Bailes
 */
public class SubPageManager implements PageManager {
    BlockFile                           file;
    /**
     * rootBlock is the root block of the entire block file, and will
     * always be zero in this implementation.
     */
    long                                rootBlock;

    /** The size of a subpage */
    int                                 pageSize;

    /** The page shift (file.pageSize >> pageShift == pageSize) */
    int                                 pageShift;

    /** The offset into the root block of our freelist head */
    int                                 offset;

    /** (1 << pageShift) */
    int                                 pagesPerBlock;

    // kinda unique.
    static final long                   PAGE_FREE       = 0xabcddbca5438182fL;

    /**
     * Construct a new sub-page manager bound to the specified block file
     */
    public SubPageManager(BlockFile file, int pageShift, long rootBlock)
        throws IOException
    {
        this.file = file;
        this.pageShift = pageShift;
        this.rootBlock = rootBlock;
        this.offset = BlockFile.oSUBPAGE_ROOT + pageShift * BlockFile.REF_SIZE;
        this.pageSize = file.getPageSize() >> pageShift;
        this.pagesPerBlock = 1 << pageShift;
        if (pageSize < BlockFile.REF_SIZE * 2) {
            throw new IOException("Sub page too small: " + pageSize);
        }
    }

    /**
     * Allocate a new block.
     */
    public long newPage() throws IOException {
        long ret = -1;
        synchronized (file.fileLock) {
            Block root = file.getBlock(rootBlock);
            Page page = null;
            try {
                ret = root.readLong(offset);
                if (ret == 0) {
                    ret = allocateNewBlock();
                }
                page = getPage(ret);
                long next = page.readLong(0);
                if (page.readLong(BlockFile.REF_SIZE) != PAGE_FREE) {
                    //#ifdef DEBUG
                    byte[] ff = new byte[pageSize];
                    page.read(0, ff, 0, pageSize);
                    Debug.println("Page: " + Util.strBytes(ff));
                    Debug.println("" + toString() + 
                                  ".newPage() " +
                                  toString(ret) +
                                  ", freelist corrupted: ");
                    //#endif
                    throw new DatafileException("" + toString() + 
                                               ".newPage() " +
                                               toString(ret) +
                                               ", freelist corrupted: ");

                }
                root.writeLong(offset, next);
                page.clear();           // Can we let caller do this?
            } finally {
                if (page != null) page.decrRefCount();
                root.decrRefCount();
            }
            //#ifdef DEBUG
            if (Trace.bit(11)) {
                Debug.println("" + toString() + ".newPage() = " +
                              toString(ret));
            }
            if (pageShift(ret) != pageShift) {
                throw new DatafileException("" + toString() + ".newPage(), " +
                                           toString(ret) +
                                           ", freelist corrupted");
            }
            //#endif
        }
        return ret;
    }
    
    /**
     * Free a block.
     *
     * @param block a previously allocated block
     */
    public void freePage(long block) throws IOException {
//-        //#ifdef PARANOID
//-        checkPage(block);
        //#endif
        //#ifdef DEBUG
        if (Trace.bit(11)) {
            Debug.println("" + toString() + ".freePage(" + 
                          toString(block) + ")");
        }
        //#endif
        if (pageShift(block) != pageShift) {
            throw new DatafileException("" + toString() + ".freePage(" +
                                       toString(block) + "), bad page");
        }
        synchronized (file.fileLock) {
            Block root = file.getBlock(rootBlock);
            try {
                long head = root.readLong(offset);
                if (head == block) {
                    throw new DatafileException("" + toString() + ",freePage(" +
                                               toString(block) + "), already" +
                                               " free");
                }
                Page page = getPage(block);
                page.writeLong(0, head);
                if (page.readLong(BlockFile.REF_SIZE) == PAGE_FREE) {
                    throw new DatafileException("" + toString() + ",freePage(" +
                                               toString(block) + "), already" +
                                               " free");
                }
                page.writeLong(BlockFile.REF_SIZE, PAGE_FREE);
                root.writeLong(offset, block);
                page.decrRefCount();
            } finally {
                root.decrRefCount();
            }
        }
    }

    /**
     * Allocate new block and slice it up into sub-pages. Link the sub-pages
     * together and return the head of the list.  Lock must be held!
     */
    private final long allocateNewBlock() throws IOException {
        long blk = file.newPage();
        //#ifdef DEBUG
        if (Trace.bit(11)) {
            Debug.println("" + toString() + ".allocateNewBlock: " + blk);
        }
        //#endif
        Block b = file.getBlock(blk);
        b.writeLong(0, 0);
        b.writeLong(BlockFile.REF_SIZE, PAGE_FREE);
        try {
            long p = makePageNo(blk, 0);
            for (int i = 1; i < pagesPerBlock; i++) {
                long np = makePageNo(blk, i);
                Page page = getPage(np);
                try {
                    page.writeLong(0, p);
                    page.writeLong(BlockFile.REF_SIZE, PAGE_FREE);
                    p = np;
                } finally {
                    page.decrRefCount();
                }
            }
            return p;
        } finally {
            b.decrRefCount();
        }
    }

    /**
     * Sub-page numbers are encoded as follows:<p>
     * <pre>
     *     sub_page {
     *        pageShift         bit4
     *        pageNum           bit16
     *        blockNum          bit44
     *     }
     * </pre><p>
     *
     * This has the desirable property of being backward compatible with the
     * pre-sub-page block numbers, which will have pageShift and pageNum
     * equal to zero.
     */

    static int[] PAGE_SHIFT = {63,4};
    static int[] PAGE_NUM   = {59,16};
//-    //#ifdef PARANOID
//-    static int[] MAGIC = {43,4};
//-    static int[] BLOCK_NUM  = {39,40};
    //#else
    static int[] BLOCK_NUM  = {43,44};
    //#endif

    static final long fIns(long v, int[] f) {
        return (v & fMask(f)) << fEnd(f);
    }
    static final long fMask(int[] f) {
        return (1L << f[1]) - 1;
    }
    static final long fEnd(int[] f) {
        return f[0] - f[1] + 1;
    }
    static final long fExt(long v, int[] f) {
        return (v >> fEnd(f)) & fMask(f);
    }

    final long makePageNo(long blk, long page) {
        // we could mask here to "prevent" errors, but that really doesnt'
        // help.  We could range check 'blk' and 'page' but we want this
        // fn to be fast, and hope we can be assured that the rest of the
        // code ensures the invariants.
        long ret = fIns(pageShift, PAGE_SHIFT)
//-            //#ifdef PARANOID
//-            | fIns(0xc, MAGIC)
            //#endif
            | fIns(page, PAGE_NUM)
            | fIns(blk, BLOCK_NUM);
//-        //#ifdef PARANOID
//-        if (fExt(ret, BLOCK_NUM) != blk) {
//-            Debug.println("pageShift: " + pageShift + ": " +
//-                          fIns(pageShift, PAGE_SHIFT));
//-            Debug.println("page: " + page + ": " +
//-                          fIns(page, PAGE_NUM));
//-            Debug.println("block: " + blk + ": " +
//-                          fIns(blk, BLOCK_NUM));
//-            Debug.println("============Bad blk: " + blk + ", " + page +
//-                          ", " + toString(ret));
//-        }
        //#endif
        return ret;
    }

    /**
     * Return the number of the block that contains this page.
     */
    public static final long pageBlock(long page) {
//-        //#ifdef PARANOID
//-        checkPage(page);
        //#endif
        return fExt(page, BLOCK_NUM);
    }

//-    //#ifdef PARANOID
//-    static public void checkPage(long page) {
//-        if (fExt(page, PAGE_SHIFT) != 0 &&
//-            (fExt(page, MAGIC) != 0xc ||
//-             (fExt(page, PAGE_NUM) >= (1 << fExt(page, PAGE_SHIFT))))) {
//-            String s = "Page[ps:" + fExt(page, PAGE_SHIFT) +
//-                " blk:" + fExt(page, BLOCK_NUM) +
//-                " off:" + fExt(page, PAGE_NUM) + ']';
//-            throw new RuntimeException("Bad page: " + s);
//-        }
//-    }
    //#endif
    
    /**
     * Return the byte offset of this page in the base block.
     */
    public final int pageOffset(long page) {
//-        //#ifdef PARANOID
//-        checkPage(page);
        //#endif
        return (int)(fExt(page, PAGE_NUM) * pageSize);
    }

    /**
     * Return the page shift for this page.
     */
    static final int pageShift(long page) {
//-        //#ifdef PARANOID
//-        checkPage(page);
        //#endif
        return (int)(fExt(page, PAGE_SHIFT));
    }
    
    /**
     * Return the specified page
     */
    public Page getPage(long block) throws IOException {
//-        //#ifdef PARANOID
//-        checkPage(block);
        //#endif
        return new SubPage(this, block);
    }

    /**
     * Return the (file) lock
     */
    public Object getLock() { return file.getLock(); }

    /**
     * Return this manager's block size
     */
    public final int getPageSize() {
        return pageSize;
    }

    /**
     * Return a new input stream, reading from the region with the specified
     * root block.
     *
     * @param block the root block of the region
     * @return an InputStream bound to the region.
     * @exception IOException if the block number isn't valid, or if another
     *          error is detected trying to access the region.
     */
    public RandomAccessInputStream getInputStream(long block) 
        throws IOException 
    {
        return new RandomAccessInputStream(getStream(block));
    }

    /**
     * Return a new output stream, writing to the region with the specified
     * root block.
     *
     * @param block the root block of the region
     * @return an OutputStream bound to the region.
     * @exception IOException if the block number isn't valid, or if another
     *          error is detected trying to access the region.
     */
    public RandomAccessOutputStream getOutputStream(long block) 
        throws IOException 
    {
        return new RandomAccessOutputStream(getStream(block));
    }

    public RandomAccess getStream(long blockRef) throws IOException {
        return new BlockAccess(this, blockRef);
    }

    /**
     * Destroy the stream with the specified root page and free up the
     * storage it was using.
     *
     * @param page the root page of the region
     *
     * @exception IOException if the page number isn't valid, or if another
     *          error is detected trying to access the region.
     */
    public void freeStream(long page) throws IOException {
        synchronized (file.fileLock) {
            getStream(page).resize(0);
            freePage(page);
        }
    }

    //#ifdef DEBUG
    /**
     * Return a displayable representation for this object.
     */
    public String toString() {
        return "PM" + pageSize + ":[" + rootBlock + "," + offset + "]";
    }
    //#endif

    public static String toString(long page) {
        int psize = 8192 >> pageShift(page);
        long pnum = fExt(page, PAGE_NUM);
        return "Page(" + psize +
            ")[" + pageBlock(page) +
            ":" + pnum + " (" + (pnum * psize) + ")" +
            ']';
    }
}