package org.dbxml.core.filer;

/*
 * dbXML License, Version 1.0
 *
 *
 * Copyright (c) 1999-2001 The dbXML Group, L.L.C.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *
 * 3. The end-user documentation included with the redistribution,
 *    if any, must include the following acknowledgment:
 *       "This product includes software developed by
 *        The dbXML Group (http://www.dbxml.com/)."
 *    Alternately, this acknowledgment may appear in the software
 *    itself, if and wherever such third-party acknowledgments normally
 *    appear.
 *
 * 4. The names "dbXML" and "The dbXML Group" must not be used to
 *    endorse or promote products derived from this software without
 *    prior written permission. For written permission, please contact
 *    info@dbxml.com.
 *
 * 5. Products derived from this software may not be called "dbXML",
 *    nor may "dbXML" appear in their name, without prior written
 *    permission of The dbXML Group.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED.  IN NO EVENT SHALL THE DBXML GROUP OR ITS CONTRIBUTORS
 * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
 * OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT
 * OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
 * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * $Id: Paged.java,v 1.18 2001/07/10 00:52:22 tom Exp $
 */

import org.dbxml.server.util.*;
import org.dbxml.server.*;
import org.dbxml.core.*;
import org.dbxml.core.data.*;

import java.io.*;
import java.util.*;

/**
 * Paged is a paged file foundation that is used by both the BTree
 * class and the HashFiler.  It provides flexible paged I/O and
 * page caching functionality.
 */

public abstract class Paged {
   private static final int MAXLOCKS = 8; // This is per Page, not File
   private static final int CACHESIZE = 131072; // 128k

   protected static final byte UNUSED = 0;
   protected static final byte OVERFLOW = 126;
   protected static final byte DELETED = 127;

   private File file;
   private boolean opened = false;
   private FileHeader fileHeader;
   private ObjectPool descriptors;
   private Map pages = new WeakHashMap(); // Long to Page

   public Paged() {
      fileHeader = createFileHeader();
   }

   public Paged(File file) {
      this();
      setFile(file);
   }

   /**
    * setFile sets the file object for this Paged.
    *
    * @param file The File
    */
   protected final void setFile(final File file) {
      this.file = file;
      descriptors = new ObjectPool() {
         public Poolable createObject() {
            try {
               return new RandomAccessPoolable(file, "rw");
            }
            catch ( Exception e ) {
               return null;
            }
         }
      };
   }

   /**
    * getFile returns the file object for this Paged.
    *
    * @return The File
    */
   protected final File getFile() {
      return file;
   }

   /**
    * getPage returns the page specified by pageNum.
    *
    * @param lp The Page number
    * @return The requested Page
    * @throws IOException if an Exception occurs
    */
   protected final Page getPage(Long lp) throws IOException {
      Page p = (Page)pages.get(lp);
      if ( p == null ) {
         p = new Page(lp.longValue());
         try {
            p.acquireReadLock();
            p.read();
            pages.put(lp, p);
         }
         finally {
            p.releaseReadLock();
         }
      }
      return p;
   }
   
   /**
    * getPage returns the page specified by pageNum.
    *
    * @param pageNum The Page number
    * @return The requested Page
    * @throws IOException if an Exception occurs
    */
   protected final Page getPage(long pageNum) throws IOException {
      return getPage(new Long(pageNum));
   }

   /**
    * readValue reads the multi-Paged Value starting at the specified
    * Page.
    *
    * @param page The starting Page
    * @return The Value
    * @throws IOException if an Exception occurs
    */
   protected final Value readValue(Page page) throws IOException {
      PageHeader sph = page.getPageHeader();
      ByteArrayOutputStream bos = new ByteArrayOutputStream((int)sph.getRecordLen());
      Page p = page;
      PageHeader ph = null;
      long nextPage;
      while ( true ) {
         ph = p.getPageHeader();
         p.streamTo(bos);
         nextPage = ph.getNextPage();
         if ( nextPage != -1 )
            p = getPage(nextPage);
         else
            break;
      }
      return new Value(bos.toByteArray());
   }

   /**
    * readValue reads the multi-Paged Value starting at the specified
    * page number.
    *
    * @param page The starting page number
    * @return The Value
    * @throws IOException if an Exception occurs
    */
   protected final Value readValue(long page) throws IOException {
      return readValue(getPage(page));
   }

   /**
    * writeValue writes the multi-Paged Value starting at the specified
    * Page.
    *
    * @param page The starting Page
    * @param value The Value to write
    * @throws IOException if an Exception occurs
    */
   protected final void writeValue(Page page, Value value) throws IOException {
      InputStream is = value.getInputStream();

      PageHeader hdr = page.getPageHeader();
      hdr.setRecordLen(value.getLength());
      page.streamFrom(is);

      // Write out the rest of the value
      while ( is.available() > 0 ) {
         Page lpage = page;
         PageHeader lhdr = hdr;

         long np = lhdr.getNextPage();
         if ( np != -1 )
            page = getPage(np);
         else {
            page = getFreePage();
            lhdr.setNextPage(page.getPageNum());
         }

         hdr = page.getPageHeader();
         hdr.setStatus(OVERFLOW);
         page.streamFrom(is);

         lpage.write();
      }

      long np = hdr.getNextPage();
      if ( np != -1 )
         unlinkPages(np);
      hdr.setNextPage(-1);
      page.write();
   }

   /**
    * writeValue writes the multi-Paged Value starting at the specified
    * page number.
    *
    * @param page The starting page number
    * @param value The Value to write
    * @throws IOException if an Exception occurs
    */
   protected final void writeValue(long page, Value value) throws IOException {
      writeValue(getPage(page), value);
   }

   /**
    * unlinkPages unlinks a set of pages starting at the specified Page.
    *
    * @param page The starting Page to unlink
    * @throws IOException if an Exception occurs
    */
   protected final void unlinkPages(Page page) throws IOException {
      // If the page is in primary space, just reset it's
      // status.  If it's in overflow, add it to the unused list.
      if ( page.pageNum < fileHeader.pageCount ) {
         long nextPage = page.header.nextPage;
         page.header.setStatus(DELETED);
         page.header.setNextPage(-1);
         page.write();
         page = nextPage != -1 ? getPage(nextPage)
                               : null;
      }

      if ( page != null ) {
         // Walk the chain and add it to the unused list
         long firstPage = page.pageNum;
         while ( page.header.nextPage != -1 )
            page = getPage(page.header.nextPage);
         long lastPage = page.pageNum;

         synchronized(fileHeader) {
            if ( fileHeader.lastFreePage != -1 ) {
               Page p = getPage(fileHeader.lastFreePage);
               p.header.setNextPage(firstPage);
               p.write();
            }
            if ( fileHeader.firstFreePage == -1 )
               fileHeader.setFirstFreePage(firstPage);
            fileHeader.setLastFreePage(lastPage);
         }
      }
   }

   /**
    * unlinkPages unlinks a set of pages starting at the specified
    * page number.
    *
    * @param page The starting page number to unlink
    * @throws IOException if an Exception occurs
    */
   protected final void unlinkPages(long pageNum) throws IOException {
      unlinkPages(getPage(pageNum));
   }

   /**
    * getFreePage returns the first free Page from secondary storage.
    * If no Pages are available, the file is grown as appropriate.
    *
    * @return The next free Page
    * @throws IOException if an Exception occurs
    */
   protected final Page getFreePage() throws IOException {
      Page p = null;
      synchronized(fileHeader) {
         long pageNum = fileHeader.firstFreePage;
         if ( pageNum != -1 ) {
            // Steal a deleted page
            p = getPage(pageNum);
            fileHeader.setFirstFreePage(p.getPageHeader().nextPage);
            if ( fileHeader.firstFreePage == -1 )
               fileHeader.setLastFreePage(-1);
         }
         else {
            // Grow the file
            pageNum = fileHeader.totalCount;
            fileHeader.setTotalCount(pageNum+1);
            p = getPage(pageNum);
         }
      }
      // Initialize The Page Header (Cleanly)
      p.header.setNextPage(-1);
      p.header.setStatus(UNUSED);
      return p;
   }

   protected final void checkOpened() throws DBException {
      if ( !opened )
         throw new FilerException(FaultCodes.COL_COLLECTION_CLOSED, "Filer is closed");
   }
   
   /**
    * getFileHeader returns the FileHeader
    *
    * @return The FileHeader
    */
   public FileHeader getFileHeader() {
      return fileHeader;
   }

   public boolean exists() {
      return file.exists();
   }

   public boolean create() throws DBException {
      try {
         fileHeader.write();
         return true;
      }
      catch ( Exception e ) {
         throw new FilerException(FaultCodes.GEN_CRITICAL_ERROR, "Error creating "+file.getName());
      }
   }

   public boolean open() throws DBException {
      try {
         if ( exists() ) {
            fileHeader.read();
            opened = true;
         }
         else
            opened = false;
      }
      catch ( Exception e ) {
         throw new FilerException(FaultCodes.GEN_CRITICAL_ERROR, "Error opening "+file.getName());
      }
      return opened;
   }

   public boolean close() throws DBException {
      opened = false;
      return true;
   }

   public boolean isOpened() {
      return opened;
   }

   public boolean drop() throws DBException {
      try {
         close();
         if ( exists() )
            return getFile().delete();
         else
            return true;
      }
      catch ( Exception e ) {
         throw new FilerException(FaultCodes.COL_CANNOT_DROP, "Can't drop Filer");
      }
   }

   /**
    * createFileHeader must be implemented by a Paged implementation
    * in order to create an appropriate subclass instance of a FileHeader.
    *
    * @return a new FileHeader
    */
   public abstract FileHeader createFileHeader();

   /**
    * createFileHeader must be implemented by a Paged implementation
    * in order to create an appropriate subclass instance of a FileHeader.
    *
    * @param read If true, reads the FileHeader from disk
    * @return a new FileHeader
    * @throws IOException if an exception occurs
    */
   public abstract FileHeader createFileHeader(boolean read) throws IOException;

   /**
    * createFileHeader must be implemented by a Paged implementation
    * in order to create an appropriate subclass instance of a FileHeader.
    *
    * @param pageCount The number of pages to allocate for primary storage
    * @return a new FileHeader
    */
   public abstract FileHeader createFileHeader(long pageCount);

   /**
    * createFileHeader must be implemented by a Paged implementation
    * in order to create an appropriate subclass instance of a FileHeader.
    *
    * @param pageCount The number of pages to allocate for primary storage
    * @param pageSize The size of a Page (should be a multiple of a FS block)
    * @return a new FileHeader
    */
   public abstract FileHeader createFileHeader(long pageCount, int pageSize);

   /**
    * createPageHeader must be implemented by a Paged implementation
    * in order to create an appropriate subclass instance of a PageHeader.
    *
    * @return a new PageHeader
    */
   public abstract PageHeader createPageHeader();


   // These are a bunch of utility methods for subclasses

   public static Value[] insertArrayValue(Value[] vals, Value val, int idx) {
      Value[] newVals = new Value[vals.length+1];
      if ( idx > 0 )
         System.arraycopy(vals, 0, newVals, 0, idx);
      newVals[idx] = val;
      if ( idx < vals.length )
         System.arraycopy(vals, idx, newVals, idx+1, vals.length-idx);
      return newVals;
   }

   public static Value[] deleteArrayValue(Value[] vals, int idx) {
      Value[] newVals = new Value[vals.length-1];
      if ( idx > 0 )
         System.arraycopy(vals, 0, newVals, 0, idx);
      if ( idx < newVals.length)
         System.arraycopy(vals, idx+1, newVals, idx, newVals.length-idx);
      return newVals;
   }

   public static long[] insertArrayLong(long[] vals, long val, int idx) {
      long[] newVals = new long[vals.length+1];
      if ( idx > 0 )
         System.arraycopy(vals, 0, newVals, 0, idx);
      newVals[idx] = val;
      if ( idx < vals.length )
         System.arraycopy(vals, idx, newVals, idx+1, vals.length-idx);
      return newVals;
   }

   public static long[] deleteArrayLong(long[] vals, int idx) {
      long[] newVals = new long[vals.length-1];
      if ( idx > 0 )
         System.arraycopy(vals, 0, newVals, 0, idx);
      if ( idx < newVals.length)
         System.arraycopy(vals, idx+1, newVals, idx, newVals.length-idx);
      return newVals;
   }

   public static int[] insertArrayInt(int[] vals, int val, int idx) {
      int[] newVals = new int[vals.length+1];
      if ( idx > 0 )
         System.arraycopy(vals, 0, newVals, 0, idx);
      newVals[idx] = val;
      if ( idx < vals.length )
         System.arraycopy(vals, idx, newVals, idx+1, vals.length-idx);
      return newVals;
   }

   public static int[] deleteArrayInt(int[] vals, int idx) {
      int[] newVals = new int[vals.length-1];
      if ( idx > 0 )
         System.arraycopy(vals, 0, newVals, 0, idx);
      if ( idx < newVals.length)
         System.arraycopy(vals, idx+1, newVals, idx, newVals.length-idx);
      return newVals;
   }

   public static short[] insertArrayShort(short[] vals, short val, int idx) {
      short[] newVals = new short[vals.length+1];
      if ( idx > 0 )
         System.arraycopy(vals, 0, newVals, 0, idx);
      newVals[idx] = val;
      if ( idx < vals.length )
         System.arraycopy(vals, idx, newVals, idx+1, vals.length-idx);
      return newVals;
   }

   public static short[] deleteArrayShort(short[] vals, int idx) {
      short[] newVals = new short[vals.length-1];
      if ( idx > 0 )
         System.arraycopy(vals, 0, newVals, 0, idx);
      if ( idx < newVals.length)
         System.arraycopy(vals, idx+1, newVals, idx, newVals.length-idx);
      return newVals;
   }


   /**
    * RandomAccessPoolable
    */

   private class RandomAccessPoolable extends RandomAccessFile implements Poolable {
      private ObjectPool pool;

      public RandomAccessPoolable(File file, String mode) throws IOException, FileNotFoundException {
         super(file, mode);
      }

      public RandomAccessPoolable(String file, String mode) throws IOException, FileNotFoundException {
         super(file, mode);
      }

      public void setPool(ObjectPool pool) {
         this.pool = pool;
      }

      public void reclaim() {
         pool.putObject(this);
      }
   }


   /**
    * FileHeader
    */

   public abstract class FileHeader {
      private boolean dirty = false;
      private int workSize;

      private short headerSize;
      private int pageSize;
      private long pageCount;
      private long totalCount;
      private long firstFreePage = -1;
      private long lastFreePage = -1;
      private byte pageHeaderSize = 64;
      private short maxKeySize = 256;
      private long recordCount;

      public FileHeader() {
         this(1024);
      }

      public FileHeader(long pageCount) {
         this(pageCount, 4096);
      }

      public FileHeader(long pageCount, int pageSize) {
         this.pageSize = pageSize;
         this.pageCount = pageCount;
         totalCount = pageCount;
         headerSize = (short)4096;
         calculateWorkSize();
      }

      public FileHeader(boolean read) throws IOException {
         if ( read )
            read();
      }

      public final synchronized void read() throws IOException {
         RandomAccessPoolable f = (RandomAccessPoolable)descriptors.getObject();
         try {
            f.seek(0);
            read(f);
            calculateWorkSize();
         }
         finally {
            f.reclaim();
         }
      }

      public void read(RandomAccessFile raf) throws IOException {
         headerSize = raf.readShort();
         pageSize = raf.readInt();
         pageCount = raf.readLong();
         totalCount = raf.readLong();
         firstFreePage = raf.readLong();
         lastFreePage = raf.readLong();
         pageHeaderSize = raf.readByte();
         maxKeySize = raf.readShort();
         recordCount = raf.readLong();
      }

      public final synchronized void write() throws IOException {
         if ( !dirty )
            return;
         
         RandomAccessPoolable f = (RandomAccessPoolable)descriptors.getObject();
         try {
            f.seek(0);
            write(f);
            dirty = false;
         }
         finally {
            f.reclaim();
         }
      }

      public void write(RandomAccessFile raf) throws IOException {
         raf.writeShort(headerSize);
         raf.writeInt(pageSize);
         raf.writeLong(pageCount);
         raf.writeLong(totalCount);
         raf.writeLong(firstFreePage);
         raf.writeLong(lastFreePage);
         raf.writeByte(pageHeaderSize);
         raf.writeShort(maxKeySize);
         raf.writeLong(recordCount);
      }

      public final void setDirty() {
         dirty = true;
      }

      public final boolean isDirty() {
         return dirty;
      }

      /** The size of the FileHeader.  Usually 1 OS Page */
      public final void setHeaderSize(short headerSize) {
         this.headerSize = headerSize;
         dirty = true;
      }

      /** The size of the FileHeader.  Usually 1 OS Page */
      public final short getHeaderSize() {
         return headerSize;
      }

      /** The size of a page.  Usually a multiple of a FS block */
      public final void setPageSize(int pageSize) {
         this.pageSize = pageSize;
         calculateWorkSize();
         dirty = true;
      }

      /** The size of a page.  Usually a multiple of a FS block */
      public final int getPageSize() {
         return pageSize;
      }

      /** The number of pages in primary storage */
      public final void setPageCount(long pageCount) {
         this.pageCount = pageCount;
         dirty = true;
      }

      /** The number of pages in primary storage */
      public final long getPageCount() {
         return pageCount;
      }

      /** The number of total pages in the file */
      public final void setTotalCount(long totalCount) {
         this.totalCount = totalCount;
         dirty = true;
      }

      /** The number of total pages in the file */
      public final long getTotalCount() {
         return totalCount;
      }

      /** The first free page in unused secondary space */
      public final void setFirstFreePage(long firstFreePage) {
         this.firstFreePage = firstFreePage;
         dirty = true;
      }

      /** The first free page in unused secondary space */
      public final long getFirstFreePage() {
         return firstFreePage;
      }

      /** The last free page in unused secondary space */
      public final void setLastFreePage(long lastFreePage) {
         this.lastFreePage = lastFreePage;
         dirty = true;
      }

      /** The last free page in unused secondary space */
      public final long getLastFreePage() {
         return lastFreePage;
      }

      /** The size of a page header. 64 is sufficient */
      public final void setPageHeaderSize(byte pageHeaderSize) {
         this.pageHeaderSize = pageHeaderSize;
         calculateWorkSize();
         dirty = true;
      }

      /** The size of a page header. 64 is sufficient */
      public final byte getPageHeaderSize() {
         return pageHeaderSize;
      }

      /** The maximum number of bytes a key can be.  256 is good */
      public final void setMaxKeySize(short maxKeySize) {
         this.maxKeySize = maxKeySize;
         dirty = true;
      }

      /** The maximum number of bytes a key can be.  256 is good */
      public final short getMaxKeySize() {
         return maxKeySize;
      }

      /** The number of records being managed by the file (not pages) */
      public final void setRecordCount(long recordCount) {
         this.recordCount = recordCount;
         dirty = true;
      }

      /** Increment the number of records being managed by the file */
      public final synchronized void incRecordCount() {
         recordCount++;
         dirty = true;
      }

      /** Decrement the number of records being managed by the file */
      public final synchronized void decRecordCount() {
         recordCount--;
         dirty = true;
      }

      /** The number of records being managed by the file (not pages) */
      public final long getRecordCount() {
         return recordCount;
      }

      private void calculateWorkSize() {
         workSize = pageSize - pageHeaderSize;
      }

      public final int getWorkSize() {
         return workSize;
      }
   }

   /**
    * PageHeader
    */

   public abstract class PageHeader implements Streamable {
      private boolean dirty = false;
      private byte status = UNUSED;
      private short keyLen = 0;
      private int keyHash = 0;
      private int dataLen = 0;
      private int recordLen = 0;
      private long nextPage = -1;

      public PageHeader() {
      }

      public PageHeader(DataInputStream dis) throws IOException {
         read(dis);
      }

      public void read(DataInputStream dis) throws IOException {
         status = dis.readByte();
         dirty = false;
         
         if ( status == UNUSED )
            return;

         keyLen = dis.readShort();
         keyHash = dis.readInt();
         dataLen = dis.readInt();
         recordLen = dis.readInt();
         nextPage = dis.readLong();
      }

      public void write(DataOutputStream dos) throws IOException {
         dirty = false;
         dos.writeByte(status);
         dos.writeShort(keyLen);
         dos.writeInt(keyHash);
         dos.writeInt(dataLen);
         dos.writeInt(recordLen);
         dos.writeLong(nextPage);
      }

      public final boolean isDirty() {
         return dirty;
      }
      
      public final void setDirty() {
         dirty = true;
      }
      
      /** The status of this page (UNUSED, RECORD, DELETED, etc...) */
      public final void setStatus(byte status) {
         this.status = status;
         dirty = true;
      }

      /** The status of this page (UNUSED, RECORD, DELETED, etc...) */
      public final byte getStatus() {
         return status;
      }

      public final void setKey(Key key) {
         // setKey WIPES OUT the Page data
         setRecordLen(0);
         dataLen = 0;
         keyHash = key.getHash();
         keyLen = (short)key.getLength();
         dirty = true;
      }

      /** The length of the Key */
      public final void setKeyLen(short keyLen) {
         this.keyLen = keyLen;
         dirty = true;
      }

      /** The length of the Key */
      public final short getKeyLen() {
         return keyLen;
      }

      /** The hashed value of the Key for quick comparisons */
      public final void setKeyHash(int keyHash) {
         this.keyHash = keyHash;
         dirty = true;
      }

      /** The hashed value of the Key for quick comparisons */
      public final int getKeyHash() {
         return keyHash;
      }

      /** The length of the Data */
      public final void setDataLen(int dataLen) {
         this.dataLen = dataLen;
         dirty = true;
      }

      /** The length of the Data */
      public final int getDataLen() {
         return dataLen;
      }

      /** The length of the Record's value */
      public void setRecordLen(int recordLen) {
         this.recordLen = recordLen;
         dirty = true;
      }

      /** The length of the Record's value */
      public final int getRecordLen() {
         return recordLen;
      }

      /** The next page for this Record (if overflowed) */
      public final void setNextPage(long nextPage) {
         this.nextPage = nextPage;
         dirty = true;
      }

      /** The next page for this Record (if overflowed) */
      public final long getNextPage() {
         return nextPage;
      }
   }

   /**
    * Page
    */

   public final class Page implements Comparable {
      /** This page number */
      private long pageNum;

      /** The data for this page */
      private byte[] data;

      /** The Header for this Page */
      private PageHeader header = createPageHeader();

      /** The position (relative) of the Key in the data array */
      private int keyPos;

      /** The position (relative) of the Data in the data array */
      private int dataPos;

      /** The offset into the file that this page starts */
      private long offset;

      /** The number of outstanding locks on the page */
      private int locks = 0;

      public Page() {
      }

      public Page(long pageNum) throws IOException {
         this();
         setPageNum(pageNum);
      }

      public synchronized void read() throws IOException {
         Page page = null;
         data = new byte[fileHeader.pageSize];
         RandomAccessPoolable f = (RandomAccessPoolable)descriptors.getObject();
         try {
            f.seek(offset);
            f.read(data);
            
            ByteArrayInputStream bis = new ByteArrayInputStream(data);
            DataInputStream dis = new DataInputStream(bis);
   
            // Read in the header
            header.read(dis);
   
            keyPos = fileHeader.pageHeaderSize;
            dataPos = keyPos + header.keyLen;
         }
         finally {
            f.reclaim();
         }
      }

      public synchronized void write() throws IOException {
         if ( !header.dirty )
            return;
         
         // Write out the header
         ByteArrayOutputStream bos = new ByteArrayOutputStream(fileHeader.getPageHeaderSize());
         DataOutputStream dos = new DataOutputStream(bos);
         header.write(dos);
         byte[] b = bos.toByteArray();
         System.arraycopy(b, 0, data, 0, b.length);

         RandomAccessPoolable f = (RandomAccessPoolable)descriptors.getObject();
         try {
            if ( offset >= f.length() ) {
               // Grow the file
               long o = (fileHeader.headerSize + ((fileHeader.totalCount*3)/2)*fileHeader.pageSize)+(fileHeader.pageSize-1);
               f.seek(o);
               f.writeByte(0);
            }
            f.seek(offset);
            f.write(data);
         }
         finally {
            f.reclaim();
         }
      }

      public synchronized void acquireReadLock() {
         while ( locks >= MAXLOCKS ) {
            try {
               wait();
            }
            catch ( Exception e ) {
               org.dbxml.Debug.printStackTrace(e);
            }
         }
         locks++;
      }

      public synchronized void releaseReadLock() {
         locks--;
         notify();
      }

      public synchronized void acquireWriteLock() {
         int total = 0;
         int stolen = 0;

         while ( total < MAXLOCKS && locks >= MAXLOCKS ) {
            try {
               wait();
            }
            catch ( Exception e ) {
               org.dbxml.Debug.printStackTrace(e);
            }
            stolen = MAXLOCKS-locks;
            total += stolen;
            locks += stolen;
         }
      }

      public synchronized void releaseWriteLock() {
         locks = 0;
         notify();
      }

      public void setPageNum(long pageNum) {
         this.pageNum = pageNum;
         offset = fileHeader.headerSize + (pageNum * fileHeader.pageSize);
      }

      public long getPageNum() {
         return pageNum;
      }

      public PageHeader getPageHeader() {
         return header;
      }

      public void setKey(Key key) {
         header.setKey(key);
         key.copyTo(data, keyPos);
      }

      public Key getKey() {
         if ( header.keyLen > 0 )
            return new Key(data, keyPos, header.keyLen);
         else
            return null;
      }

      public void streamTo(OutputStream os) throws IOException {
         if ( header.dataLen > 0 )
            os.write(data, dataPos, header.dataLen);
      }

      public void streamFrom(InputStream is) throws IOException {
         int avail = is.available();
         header.dataLen = fileHeader.workSize - header.keyLen;
         if ( avail < header.dataLen )
            header.dataLen = avail;
         if ( header.dataLen > 0 )
            is.read(data, keyPos + header.keyLen, header.dataLen);
      }
      
      public int compareTo(Object o) {
         return (int)(pageNum - ((Page)o).pageNum);
      }
   }
}

