store.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 1,996 行 · 第 1/4 页

  /**
   * Allocates a new block for a row.
   *
   * @return the block id of the allocated block.
   */
  public Block allocateRow()
    throws IOException
  {
    return allocateBlock(ALLOC_ROW);
  }

  /**
   * Return true if the block is a row block.
   */
  public boolean isRowBlock(long blockAddress)
  {
    return getAllocation(blockAddress / BLOCK_SIZE) == ALLOC_ROW;
  }

  /**
   * Allocates a new block for a non-row.
   *
   * @return the block id of the allocated block.
   */
  public Block allocateBlock()
    throws IOException
  {
    return allocateBlock(ALLOC_USED);
  }

  /**
   * Allocates a new block for a fragment
   *
   * @return the block id of the allocated block.
   */
  private Block allocateFragmentBlock()
    throws IOException
  {
    return allocateBlock(ALLOC_FRAGMENT);
  }

  /**
   * Allocates a new block for a mini-fragment
   *
   * @return the block id of the allocated block.
   */
  private Block allocateMiniFragmentBlock()
    throws IOException
  {
    return allocateBlock(ALLOC_MINI_FRAG);
  }

  /**
   * Allocates a new block for an index
   *
   * @return the block id of the allocated block.
   */
  public Block allocateIndexBlock()
    throws IOException
  {
    return allocateBlock(ALLOC_INDEX);
  }

  /**
   * Return true if the block is an index block.
   */
  public boolean isIndexBlock(long blockAddress)
  {
    return getAllocation(blockAddress / BLOCK_SIZE) == ALLOC_INDEX;
  }

  /**
   * Allocates a new block.
   *
   * @return the block id of the allocated block.
   */
  private Block allocateBlock(int code)
    throws IOException
  {
    long blockIndex;
    boolean isFileExtended = false;

    synchronized (_allocationLock) {
      long end = _blockCount;

      if (_allocationTable.length < ALLOC_BYTES_PER_BLOCK * end)
	end = _allocationTable.length / ALLOC_BYTES_PER_BLOCK;

      for (blockIndex = 0; blockIndex < end; blockIndex++) {
	if (getAllocation(blockIndex) == ALLOC_FREE)
	  break;
      }

      if (_allocationTable.length <= ALLOC_BYTES_PER_BLOCK * blockIndex) {
	// expand the allocation table
	byte []newTable = new byte[_allocationTable.length + ALLOC_CHUNK_SIZE];
	System.arraycopy(_allocationTable, 0,
			 newTable, 0,
			 _allocationTable.length);
	_allocationTable = newTable;

	// if the allocation table is over 32k, allocate the block for the
	// extension (each allocation block of 32k allocates 2G)
	if (blockIndex % (BLOCK_SIZE / ALLOC_BYTES_PER_BLOCK) == 0) {
	  setAllocation(blockIndex, ALLOC_USED);
	  blockIndex++;
	}
      }

      // mark USED before actual code so it's properly initialized
      setAllocation(blockIndex, ALLOC_USED);

      if (log.isLoggable(Level.FINE))
	log.fine(this + " allocating block " + blockIndex + " " + codeToName(code));

      if (_blockCount <= blockIndex) {
	isFileExtended = true;
	_blockCount = blockIndex + 1;
      }
    }

    long blockId = blockIndexToBlockId(blockIndex);

    Block block = _blockManager.getBlock(this, blockId);

    byte []buffer = block.getBuffer();

    for (int i = BLOCK_SIZE - 1; i >= 0; i--)
      buffer[i] = 0;

    block.setDirty(0, BLOCK_SIZE);

    // if extending file, write the contents now
    if (isFileExtended) {
      try {
	block.write();
      } catch (IOException e) {
	log.log(Level.WARNING, e.toString(), e);
      }
    }

    synchronized (_allocationLock) {
      setAllocation(blockIndex, code);
    }
    
    saveAllocation();

    return block;
  }

  /**
   * Check that an allocated block is valid.
   */
  protected void validateBlockId(long blockId)
    throws IllegalArgumentException, IllegalStateException
  {
    RuntimeException e = null;
    
    if (isClosed())
      e = new IllegalStateException(L.l("store {0} is closing.", this));
    else if (getId() <= 0)
      e = new IllegalStateException(L.l("invalid store {0}.", this));
    else if (getId() != (blockId & BLOCK_INDEX_MASK)) {
      e = new IllegalArgumentException(L.l("block {0} must match store {1}.",
					     blockId & BLOCK_INDEX_MASK,
					     this));
    }

    if (e != null)
      throw e;
  }

  /**
   * Check that an allocated block is valid.
   */
  protected void assertStoreActive()
    throws IllegalStateException
  {
    RuntimeException e = null;
    
    if (isClosed())
      e = new IllegalStateException(L.l("store {0} is closing.", this));
    else if (getId() <= 0)
      e = new IllegalStateException(L.l("invalid store {0}.", this));

    if (e != null)
      throw e;
  }
  
  /**
   * Frees a block.
   *
   * @return the block id of the allocated block.
   */
  protected void freeBlock(long blockId)
    throws IOException
  {
    if (blockId == 0)
      return;
    
    synchronized (_allocationLock) {
      setAllocation(blockIdToIndex(blockId), ALLOC_FREE);
    }

    saveAllocation();
  }

  /**
   * Sets the allocation for a block.
   */
  private final int getAllocation(long blockIndex)
  {
    int allocOffset = (int) (ALLOC_BYTES_PER_BLOCK * blockIndex);

    return _allocationTable[allocOffset] & ALLOC_MASK;
  }

  /**
   * Sets the allocation for a block.
   */
  private void setAllocation(long blockIndex, int code)
  {
    int allocOffset = (int) (ALLOC_BYTES_PER_BLOCK * blockIndex);
    
    for (int i = 1; i < ALLOC_BYTES_PER_BLOCK; i++)
      _allocationTable[allocOffset + i] = 0;

    _allocationTable[allocOffset] = (byte) code;

    setAllocDirty(allocOffset, allocOffset + ALLOC_BYTES_PER_BLOCK);
  }

  /**
   * Sets the dirty range for the allocation table.
   */
  private void setAllocDirty(int min, int max)
  {
    if (min < _allocDirtyMin)
      _allocDirtyMin = min;
    
    if (_allocDirtyMax < max)
      _allocDirtyMax = max;
  }

  /**
   * Sets the allocation for a block.
   */
  void saveAllocation()
    throws IOException
  {
    // cache doesn't actually need to write this data
    if (! _isFlushDirtyBlocksOnCommit)
      return;
    
    synchronized (_allocationWriteLock) {
      int dirtyMin;
      int dirtyMax;

      synchronized (_allocationLock) {
	dirtyMin = _allocDirtyMin;
	_allocDirtyMin = Integer.MAX_VALUE;
	
	dirtyMax = _allocDirtyMax;
	_allocDirtyMax = 0;
      }

      // Write each dirty block to disk.  The physical blocks are
      // broken up each BLOCK_SIZE / ALLOC_BYTES_PER_BLOCK.
      for (;
	   dirtyMin < dirtyMax;
	   dirtyMin = (dirtyMin + BLOCK_SIZE) - dirtyMin % BLOCK_SIZE) {
	int block = dirtyMin / (BLOCK_SIZE / ALLOC_BYTES_PER_BLOCK);
	
	int offset = dirtyMin % BLOCK_SIZE;
	int length;

	if (dirtyMin / BLOCK_SIZE != dirtyMax / BLOCK_SIZE)
	  length = BLOCK_SIZE - offset;
	else
	  length = dirtyMax - dirtyMin;

	writeBlock((long) block * BLOCK_SIZE + offset,
		   _allocationTable, offset, length);
      }
    }
  }
  
  /**
   * Reads a fragment.
   *
   * @param fragmentAddress the address of the fragment
   * @param fragmentOffset the offset inside the fragment to start reading
   * @param buffer the result buffer
   * @param offset offset into the result buffer
   * @param length the number of bytes to read
   *
   * @return the number of bytes read
   */
  public int readFragment(long fragmentAddress, int fragmentOffset,
			  byte []buffer, int offset, int length)
    throws IOException
  {
    if (fragmentAddress <= 0) {
      log.warning(this + " illegal fragment read with fragment-address=0");
      return 0;
    }
    
    if (FRAGMENT_SIZE - fragmentOffset < length) {
      // server/13df
      throw new IllegalArgumentException(L.l("read offset {0} length {1} too long",
					     fragmentOffset, length));
    }

    Block block = readBlock(addressToBlockId(fragmentAddress));

    try {
      int blockOffset = getFragmentOffset(fragmentAddress);

      byte []blockBuffer = block.getBuffer();

      synchronized (blockBuffer) {
	System.arraycopy(blockBuffer, blockOffset + fragmentOffset,
			 buffer, offset, length);
      }

      return length;
    } finally {
      block.free();
    }
  }
  
  /**
   * Reads a fragment for a clob.
   *
   * @param fragmentAddress the address of the fragment
   * @param fragmentOffset the offset inside the fragment to start reading
   * @param buffer the result buffer
   * @param offset offset into the result buffer
   * @param length the length of the fragment in characters
   *
   * @return the number of characters read
   */
  public int readFragment(long fragmentAddress, int fragmentOffset,
			  char []buffer, int offset, int length)
    throws IOException
  {
    if (FRAGMENT_SIZE - fragmentOffset < 2 * length) {
      // server/13df
      throw new IllegalArgumentException(L.l("read offset {0} length {1} too long",
					     fragmentOffset, length));
    }

    Block block = readBlock(addressToBlockId(fragmentAddress));

    try {
      int blockOffset = getFragmentOffset(fragmentAddress);
      blockOffset += fragmentOffset;

      byte []blockBuffer = block.getBuffer();
      
      synchronized (blockBuffer) {
	for (int i = 0; i < length; i++) {
	  int ch1 = blockBuffer[blockOffset] & 0xff;
	  int ch2 = blockBuffer[blockOffset + 1] & 0xff;

	  buffer[offset + i] = (char) ((ch1 << 8) + ch2);

	  blockOffset += 2;
	}
      }

      return length;
    } finally {
      block.free();
    }
  }
  
  /**
   * Reads a long value from a fragment.
   *
   * @return the long value
   */
  public long readFragmentLong(long fragmentAddress,
			       int fragmentOffset)
    throws IOException
  {
    Block block = readBlock(addressToBlockId(fragmentAddress));

    try {
      int blockOffset = getFragmentOffset(fragmentAddress);

      byte []blockBuffer = block.getBuffer();

      synchronized (blockBuffer) {
	return readLong(blockBuffer, blockOffset + fragmentOffset);
      }
    } finally {
      block.free();
    }
  }
  
  /**
   * Reads a block.
   *
   * @param blockAddress the address of the block
   * @param blockOffset the offset inside the block to start reading
   * @param buffer the result buffer
   * @param offset offset into the result buffer
   * @param length the number of bytes to read
   *
   * @return the number of bytes read
   */
  public int readBlock(long blockAddress, int blockOffset,
		       byte []buffer, int offset, int length)
    throws IOException
  {
    if (BLOCK_SIZE - blockOffset < length) {
      // server/13df
      throw new IllegalArgumentException(L.l("read offset {0} length {1} too long",
					     blockOffset, length));
    }

    Block block = readBlock(addressToBlockId(blockAddress));

    try {
      byte []blockBuffer = block.getBuffer();

      synchronized (blockBuffer) {
	System.arraycopy(blockBuffer, blockOffset,
			 buffer, offset, length);
      }

      return length;
    } finally {
      block.free();
    }
  }
  
  /**
   * Reads a block for a clob.
   *
   * @param blockAddress the address of the block
   * @param blockOffset the offset inside the block to start reading
   * @param buffer the result buffer
   * @param offset offset into the result buffer
   * @param length the length of the block in characters
   *
   * @return the number of characters read
   */
  public int readBlock(long blockAddress, int blockOffset,
		       char []buffer, int offset, int length)
    throws IOException
  {
    if (BLOCK_SIZE - blockOffset < 2 * length) {
      // server/13df
      throw new IllegalArgumentException(L.l("read offset {0} length {1} too long",
					     blockOffset, length));
    }

    Block block = readBlock(addressToBlockId(blockAddress));

    try {
      byte []blockBuffer = block.getBuffer();
      
      synchronized (blockBuffer) {
	for (int i = 0; i < length; i++) {
	  int ch1 = blockBuffer[blockOffset] & 0xff;
	  int ch2 = blockBuffer[blockOffset + 1] & 0xff;

	  buffer[offset + i] = (char) ((ch1 << 8) + ch2);

	  blockOffset += 2;
	}
      }

      return length;
    } finally {
      block.free();
    }
  }
  
  /**
   * Reads a long value from a block.
   *
   * @return the long value
   */
  public long readBlockLong(long blockAddress,
			    int offset)
    throws IOException
  {
    Block block = readBlock(addressToBlockId(blockAddress));

    try {
      byte []blockBuffer = block.getBuffer();

      synchronized (blockBuffer) {
	return readLong(blockBuffer, offset);
      }