bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/LedgerEntryPage.java - bookkeeper - Git at Google

 /*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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.bookkeeper.bookie;

 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.NoSuchElementException;
 import java.util.PrimitiveIterator.OfLong;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.bookkeeper.proto.BookieProtocol;
 import org.apache.bookkeeper.util.ZeroBuffer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * This is a page in the LedgerCache. It holds the locations
  * (entrylogfile, offset) for entry ids.
  */
 public class LedgerEntryPage implements AutoCloseable {

     private static final Logger LOG = LoggerFactory.getLogger(LedgerEntryPage.class);

     private static final int indexEntrySize = 8;
     private final int pageSize;
     private final int entriesPerPage;
     private volatile EntryKey entryKey = new EntryKey(-1, BookieProtocol.INVALID_ENTRY_ID);
     private final ByteBuffer page;
     private volatile boolean clean = true;
     private final AtomicInteger useCount = new AtomicInteger(0);
     private final AtomicInteger version = new AtomicInteger(0);
     private volatile int last = -1; // Last update position
     private final LEPStateChangeCallback callback;
     private boolean deleted;

     public static int getIndexEntrySize() {
         return indexEntrySize;
     }

     public LedgerEntryPage(int pageSize, int entriesPerPage) {
         this(pageSize, entriesPerPage, null);
     }

     public LedgerEntryPage(int pageSize, int entriesPerPage, LEPStateChangeCallback callback) {
         this.pageSize = pageSize;
         this.entriesPerPage = entriesPerPage;
         page = ByteBuffer.allocateDirect(pageSize);
         this.callback = callback;
         if (null != this.callback) {
             callback.onResetInUse(this);
         }
     }

     // Except for not allocating a new direct byte buffer; this should do everything that
     // the constructor does
     public void resetPage() {
         page.clear();
         ZeroBuffer.put(page);
         last = -1;
         entryKey = new EntryKey(-1, BookieProtocol.INVALID_ENTRY_ID);
         clean = true;
         useCount.set(0);
         deleted = false;
         if (null != this.callback) {
             callback.onResetInUse(this);
         }
     }

     public void markDeleted() {
         deleted = true;
         version.incrementAndGet();
     }

     public boolean isDeleted() {
         return deleted;
     }

     @Override
     public String toString() {
         StringBuilder sb = new StringBuilder();
         sb.append(getLedger());
         sb.append('@');
         sb.append(getFirstEntry());
         sb.append(clean ? " clean " : " dirty ");
         sb.append(useCount.get());
         return sb.toString();
     }

     public void usePage() {
         int oldVal = useCount.getAndIncrement();
         if ((0 == oldVal) && (null != callback)) {
             callback.onSetInUse(this);
         }
     }

     public void releasePageNoCallback() {
         releasePageInternal(false);
     }

     public void releasePage() {
         releasePageInternal(true);
     }

     private void releasePageInternal(boolean shouldCallback) {
         int newUseCount = useCount.decrementAndGet();
         if (newUseCount < 0) {
             throw new IllegalStateException("Use count has gone below 0");
         }
         if (shouldCallback && (null != callback) && (newUseCount == 0)) {
             callback.onResetInUse(this);
         }
     }

     private void checkPage() {
         if (useCount.get() <= 0) {
             throw new IllegalStateException("Page not marked in use");
         }
     }

     @Override
     public boolean equals(Object other) {
         if (other instanceof LedgerEntryPage) {
             LedgerEntryPage otherLEP = (LedgerEntryPage) other;
             return otherLEP.getLedger() == getLedger() && otherLEP.getFirstEntry() == getFirstEntry();
         } else {
             return false;
         }
     }

     @Override
     public int hashCode() {
         return (int) getLedger() ^ (int) (getFirstEntry());
     }

     void setClean(int versionOfCleaning) {
         this.clean = (versionOfCleaning == version.get());

         if ((null != callback) && clean) {
             callback.onSetClean(this);
         }
     }

     boolean isClean() {
         return clean;
     }

     public void setOffset(long offset, int position) {
         checkPage();
         page.putLong(position, offset);
         version.incrementAndGet();
         if (last < position / getIndexEntrySize()) {
             last = position / getIndexEntrySize();
         }
         this.clean = false;

         if (null != callback) {
             callback.onSetDirty(this);
         }
     }

     public long getOffset(int position) {
         checkPage();
         return page.getLong(position);
     }

     public void zeroPage() {
         checkPage();
         page.clear();
         ZeroBuffer.put(page);
         last = -1;
         clean = true;
     }

     public void readPage(FileInfo fi) throws IOException {
         checkPage();
         page.clear();
         try {
             fi.read(page, getFirstEntryPosition(), true);
         } catch (ShortReadException sre) {
             throw new ShortReadException("Short page read of ledger " + getLedger()
                     + " tried to get " + page.capacity() + " from position "
                     + getFirstEntryPosition() + " still need " + page.remaining(), sre);
         } catch (IllegalArgumentException iae) {
             LOG.error("IllegalArgumentException when trying to read ledger {} from position {}",
                     getLedger(), getFirstEntryPosition(), iae);
             throw iae;
         }
         // make sure we don't include partial index entry
         if (page.remaining() != 0) {
             LOG.info("Short page read of ledger {} : tried to read {} bytes from position {}, but only {} bytes read.",
                     getLedger(), page.capacity(), getFirstEntryPosition(), page.position());
             if (page.position() % indexEntrySize != 0) {
                 int partialIndexEntryStart = page.position() - page.position() % indexEntrySize;
                 page.putLong(partialIndexEntryStart, 0L);
             }
         }
         last = getLastEntryIndex();
         clean = true;
     }

     public ByteBuffer getPageToWrite() {
         checkPage();
         page.clear();
         // Different callers to this method should be able to reasonably expect independent read pointers
         return page.duplicate();
     }

     long getLedger() {
         return entryKey.getLedgerId();
     }

     public int getVersion() {
         return version.get();
     }

     public EntryKey getEntryKey() {
         return entryKey;
     }

     void setLedgerAndFirstEntry(long ledgerId, long firstEntry) {
         if (firstEntry % entriesPerPage != 0) {
             throw new IllegalArgumentException(firstEntry + " is not a multiple of " + entriesPerPage);
         }
         this.entryKey = new EntryKey(ledgerId, firstEntry);
     }
     long getFirstEntry() {
         return entryKey.getEntryId();
     }

     long getMaxPossibleEntry() {
         return entryKey.getEntryId() + entriesPerPage;
     }

     long getFirstEntryPosition() {
         return entryKey.getEntryId() * indexEntrySize;
     }

     public boolean inUse() {
         return useCount.get() > 0;
     }

     private int getLastEntryIndex() {
         for (int i = entriesPerPage - 1; i >= 0; i--) {
             if (getOffset(i * getIndexEntrySize()) > 0) {
                 return i;
             }
         }
         return -1;
     }

     public long getLastEntry() {
         if (last >= 0) {
             return last + entryKey.getEntryId();
         } else {
             int index = getLastEntryIndex();
             return index >= 0 ? (index + entryKey.getEntryId()) : 0;
         }
     }

     /**
      * Interface for getEntries to propagate entry, pos pairs.
      */
     public interface EntryVisitor {
         boolean visit(long entry, long pos) throws Exception;
     }

     /**
      * Iterates over non-empty entry mappings.
      *
      * @param vis Consumer for entry position pairs.
      * @throws Exception
      */
     public void getEntries(EntryVisitor vis) throws Exception {
         // process a page
         for (int i = 0; i < entriesPerPage; i++) {
             long offset = getOffset(i * 8);
             if (offset != 0) {
                 if (!vis.visit(getFirstEntry() + i, offset)) {
                     return;
                 }
             }
         }
     }

     public OfLong getEntriesIterator() {
         return new OfLong() {
             long firstEntry = getFirstEntry();
             int curDiffEntry = 0;

             @Override
             public boolean hasNext() {
                 while ((curDiffEntry < entriesPerPage) && (getOffset(curDiffEntry * 8) == 0)) {
                     curDiffEntry++;
                 }
                 return (curDiffEntry != entriesPerPage);
             }

             @Override
             public long nextLong() {
                 if (!hasNext()) {
                     throw new NoSuchElementException();
                 }
                 long nextEntry = firstEntry + curDiffEntry;
                 curDiffEntry++;
                 return nextEntry;
             }
         };
     }

     @Override
     public void close() throws Exception {
         releasePage();
     }
 }
	/*
	*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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.bookkeeper.bookie;

	import java.io.IOException;
	import java.nio.ByteBuffer;
	import java.util.NoSuchElementException;
	import java.util.PrimitiveIterator.OfLong;
	import java.util.concurrent.atomic.AtomicInteger;
	import org.apache.bookkeeper.proto.BookieProtocol;
	import org.apache.bookkeeper.util.ZeroBuffer;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* This is a page in the LedgerCache. It holds the locations
	* (entrylogfile, offset) for entry ids.
	*/
	public class LedgerEntryPage implements AutoCloseable {

	private static final Logger LOG = LoggerFactory.getLogger(LedgerEntryPage.class);

	private static final int indexEntrySize = 8;
	private final int pageSize;
	private final int entriesPerPage;
	private volatile EntryKey entryKey = new EntryKey(-1, BookieProtocol.INVALID_ENTRY_ID);
	private final ByteBuffer page;
	private volatile boolean clean = true;
	private final AtomicInteger useCount = new AtomicInteger(0);
	private final AtomicInteger version = new AtomicInteger(0);
	private volatile int last = -1; // Last update position
	private final LEPStateChangeCallback callback;
	private boolean deleted;

	public static int getIndexEntrySize() {
	return indexEntrySize;
	}

	public LedgerEntryPage(int pageSize, int entriesPerPage) {
	this(pageSize, entriesPerPage, null);
	}

	public LedgerEntryPage(int pageSize, int entriesPerPage, LEPStateChangeCallback callback) {
	this.pageSize = pageSize;
	this.entriesPerPage = entriesPerPage;
	page = ByteBuffer.allocateDirect(pageSize);
	this.callback = callback;
	if (null != this.callback) {
	callback.onResetInUse(this);
	}
	}

	// Except for not allocating a new direct byte buffer; this should do everything that
	// the constructor does
	public void resetPage() {
	page.clear();
	ZeroBuffer.put(page);
	last = -1;
	entryKey = new EntryKey(-1, BookieProtocol.INVALID_ENTRY_ID);
	clean = true;
	useCount.set(0);
	deleted = false;
	if (null != this.callback) {
	callback.onResetInUse(this);
	}
	}

	public void markDeleted() {
	deleted = true;
	version.incrementAndGet();
	}

	public boolean isDeleted() {
	return deleted;
	}

	@Override
	public String toString() {
	StringBuilder sb = new StringBuilder();
	sb.append(getLedger());
	sb.append('@');
	sb.append(getFirstEntry());
	sb.append(clean ? " clean " : " dirty ");
	sb.append(useCount.get());
	return sb.toString();
	}

	public void usePage() {
	int oldVal = useCount.getAndIncrement();
	if ((0 == oldVal) && (null != callback)) {
	callback.onSetInUse(this);
	}
	}

	public void releasePageNoCallback() {
	releasePageInternal(false);
	}

	public void releasePage() {
	releasePageInternal(true);
	}

	private void releasePageInternal(boolean shouldCallback) {
	int newUseCount = useCount.decrementAndGet();
	if (newUseCount < 0) {
	throw new IllegalStateException("Use count has gone below 0");
	}
	if (shouldCallback && (null != callback) && (newUseCount == 0)) {
	callback.onResetInUse(this);
	}
	}

	private void checkPage() {
	if (useCount.get() <= 0) {
	throw new IllegalStateException("Page not marked in use");
	}
	}

	@Override
	public boolean equals(Object other) {
	if (other instanceof LedgerEntryPage) {
	LedgerEntryPage otherLEP = (LedgerEntryPage) other;
	return otherLEP.getLedger() == getLedger() && otherLEP.getFirstEntry() == getFirstEntry();
	} else {
	return false;
	}
	}

	@Override
	public int hashCode() {
	return (int) getLedger() ^ (int) (getFirstEntry());
	}

	void setClean(int versionOfCleaning) {
	this.clean = (versionOfCleaning == version.get());

	if ((null != callback) && clean) {
	callback.onSetClean(this);
	}
	}

	boolean isClean() {
	return clean;
	}

	public void setOffset(long offset, int position) {
	checkPage();
	page.putLong(position, offset);
	version.incrementAndGet();
	if (last < position / getIndexEntrySize()) {
	last = position / getIndexEntrySize();
	}
	this.clean = false;

	if (null != callback) {
	callback.onSetDirty(this);
	}
	}

	public long getOffset(int position) {
	checkPage();
	return page.getLong(position);
	}

	public void zeroPage() {
	checkPage();
	page.clear();
	ZeroBuffer.put(page);
	last = -1;
	clean = true;
	}

	public void readPage(FileInfo fi) throws IOException {
	checkPage();
	page.clear();
	try {
	fi.read(page, getFirstEntryPosition(), true);
	} catch (ShortReadException sre) {
	throw new ShortReadException("Short page read of ledger " + getLedger()
	+ " tried to get " + page.capacity() + " from position "
	+ getFirstEntryPosition() + " still need " + page.remaining(), sre);
	} catch (IllegalArgumentException iae) {
	LOG.error("IllegalArgumentException when trying to read ledger {} from position {}",
	getLedger(), getFirstEntryPosition(), iae);
	throw iae;
	}
	// make sure we don't include partial index entry
	if (page.remaining() != 0) {
	LOG.info("Short page read of ledger {} : tried to read {} bytes from position {}, but only {} bytes read.",
	getLedger(), page.capacity(), getFirstEntryPosition(), page.position());
	if (page.position() % indexEntrySize != 0) {
	int partialIndexEntryStart = page.position() - page.position() % indexEntrySize;
	page.putLong(partialIndexEntryStart, 0L);
	}
	}
	last = getLastEntryIndex();
	clean = true;
	}

	public ByteBuffer getPageToWrite() {
	checkPage();
	page.clear();
	// Different callers to this method should be able to reasonably expect independent read pointers
	return page.duplicate();
	}

	long getLedger() {
	return entryKey.getLedgerId();
	}

	public int getVersion() {
	return version.get();
	}

	public EntryKey getEntryKey() {
	return entryKey;
	}

	void setLedgerAndFirstEntry(long ledgerId, long firstEntry) {
	if (firstEntry % entriesPerPage != 0) {
	throw new IllegalArgumentException(firstEntry + " is not a multiple of " + entriesPerPage);
	}
	this.entryKey = new EntryKey(ledgerId, firstEntry);
	}
	long getFirstEntry() {
	return entryKey.getEntryId();
	}

	long getMaxPossibleEntry() {
	return entryKey.getEntryId() + entriesPerPage;
	}

	long getFirstEntryPosition() {
	return entryKey.getEntryId() * indexEntrySize;
	}

	public boolean inUse() {
	return useCount.get() > 0;
	}

	private int getLastEntryIndex() {
	for (int i = entriesPerPage - 1; i >= 0; i--) {
	if (getOffset(i * getIndexEntrySize()) > 0) {
	return i;
	}
	}
	return -1;
	}

	public long getLastEntry() {
	if (last >= 0) {
	return last + entryKey.getEntryId();
	} else {
	int index = getLastEntryIndex();
	return index >= 0 ? (index + entryKey.getEntryId()) : 0;
	}
	}

	/**
	* Interface for getEntries to propagate entry, pos pairs.
	*/
	public interface EntryVisitor {
	boolean visit(long entry, long pos) throws Exception;
	}

	/**
	* Iterates over non-empty entry mappings.
	*
	* @param vis Consumer for entry position pairs.
	* @throws Exception
	*/
	public void getEntries(EntryVisitor vis) throws Exception {
	// process a page
	for (int i = 0; i < entriesPerPage; i++) {
	long offset = getOffset(i * 8);
	if (offset != 0) {
	if (!vis.visit(getFirstEntry() + i, offset)) {
	return;
	}
	}
	}
	}

	public OfLong getEntriesIterator() {
	return new OfLong() {
	long firstEntry = getFirstEntry();
	int curDiffEntry = 0;

	@Override
	public boolean hasNext() {
	while ((curDiffEntry < entriesPerPage) && (getOffset(curDiffEntry * 8) == 0)) {
	curDiffEntry++;
	}
	return (curDiffEntry != entriesPerPage);
	}

	@Override
	public long nextLong() {
	if (!hasNext()) {
	throw new NoSuchElementException();
	}
	long nextEntry = firstEntry + curDiffEntry;
	curDiffEntry++;
	return nextEntry;
	}
	};
	}

	@Override
	public void close() throws Exception {
	releasePage();
	}
	}