Modified: hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/bookie/LedgerCache.java URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/bookie/LedgerCache.java?rev=944003&r1=944002&r2=944003&view=diff ============================================================================== --- hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/bookie/LedgerCache.java (original) +++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/bookie/LedgerCache.java Thu May 13 20:23:35 2010 @@ -33,6 +33,8 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Random; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import org.apache.log4j.Logger; @@ -48,6 +50,8 @@ public class LedgerCache { public LedgerCache(File ledgerDirectories[]) { this.ledgerDirectories = ledgerDirectories; + // Retrieve all of the active ledgers. + getActiveLedgers(); } /** * the list of potentially clean ledgers @@ -63,6 +67,9 @@ public class LedgerCache { LinkedList<Long> openLedgers = new LinkedList<Long>(); + // Stores the set of active (non-deleted) ledgers. + ConcurrentMap<Long, Boolean> activeLedgers = new ConcurrentHashMap<Long, Boolean>(); + static int OPEN_FILE_LIMIT = 900; static { if (System.getProperty("openFileLimit") != null) { @@ -208,6 +215,12 @@ public class LedgerCache { File dir = pickDirs(ledgerDirectories); lf = new File(dir, ledgerName); checkParents(lf); + // A new ledger index file has been created for this Bookie. + // Add this new ledger to the set of active ledgers. + if (LOG.isDebugEnabled()) { + LOG.debug("New ledger index file created for ledgerId: " + ledger); + } + activeLedgers.put(ledger, true); } if (openLedgers.size() > OPEN_FILE_LIMIT) { fileInfoCache.remove(openLedgers.removeFirst()).close(); @@ -451,4 +464,73 @@ public class LedgerCache { return lastEntry; } + + /** + * This method will look within the ledger directories for the ledger index + * files. That will comprise the set of active ledgers this particular + * BookieServer knows about that have not yet been deleted by the BookKeeper + * Client. This is called only once during initialization. + */ + private void getActiveLedgers() { + // Ledger index files are stored in a file hierarchy with a parent and + // grandParent directory. We'll have to go two levels deep into these + // directories to find the index files. + for (File ledgerDirectory : ledgerDirectories) { + for (File grandParent : ledgerDirectory.listFiles()) { + if (grandParent.isDirectory()) { + for (File parent : grandParent.listFiles()) { + if (parent.isDirectory()) { + for (File index : parent.listFiles()) { + if (!index.isFile() || !index.getName().endsWith(".idx")) { + continue; + } + // We've found a ledger index file. The file name is the + // HexString representation of the ledgerId. + String ledgerIdInHex = index.getName().substring(0, index.getName().length() - 4); + activeLedgers.put(Long.parseLong(ledgerIdInHex, 16), true); + } + } + } + } + } + } + if (LOG.isDebugEnabled()) { + LOG.debug("Active ledgers found: " + activeLedgers); + } + } + + /** + * This method is called whenever a ledger is deleted by the BookKeeper Client + * and we want to remove all relevant data for it stored in the LedgerCache. + */ + void deleteLedger(long ledgerId) throws IOException { + if (LOG.isDebugEnabled()) + LOG.debug("Deleting ledgerId: " + ledgerId); + // Delete the ledger's index file and close the FileInfo + FileInfo fi = getFileInfo(ledgerId, false); + fi.getFile().delete(); + fi.close(); + + // Remove it from the activeLedgers set + activeLedgers.remove(ledgerId); + + // Now remove it from all the other lists and maps. + // These data structures need to be synchronized first before removing entries. + synchronized(this) { + pages.remove(ledgerId); + } + synchronized(fileInfoCache) { + fileInfoCache.remove(ledgerId); + } + synchronized(cleanLedgers) { + cleanLedgers.remove(ledgerId); + } + synchronized(dirtyLedgers) { + dirtyLedgers.remove(ledgerId); + } + synchronized(openLedgers) { + openLedgers.remove(ledgerId); + } + } + }
Modified: hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/AsyncCallback.java URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/AsyncCallback.java?rev=944003&r1=944002&r2=944003&view=diff ============================================================================== --- hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/AsyncCallback.java (original) +++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/AsyncCallback.java Thu May 13 20:23:35 2010 @@ -98,4 +98,17 @@ public interface AsyncCallback { void readComplete(int rc, LedgerHandle lh, Enumeration<LedgerEntry> seq, Object ctx); } + + public interface DeleteCallback { + /** + * Callback definition for delete operations + * + * @param rc + * return code + * @param ctx + * control object + */ + void deleteComplete(int rc, Object ctx); + } + } Modified: hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/BookKeeper.java URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/BookKeeper.java?rev=944003&r1=944002&r2=944003&view=diff ============================================================================== --- hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/BookKeeper.java (original) +++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/src/java/org/apache/bookkeeper/client/BookKeeper.java Thu May 13 20:23:35 2010 @@ -23,19 +23,18 @@ package org.apache.bookkeeper.client; import java.io.IOException; import java.util.concurrent.Executors; -import org.apache.bookkeeper.client.BKException; + import org.apache.bookkeeper.client.AsyncCallback.CreateCallback; +import org.apache.bookkeeper.client.AsyncCallback.DeleteCallback; import org.apache.bookkeeper.client.AsyncCallback.OpenCallback; import org.apache.bookkeeper.client.BKException.Code; -import org.apache.bookkeeper.client.SyncCounter; import org.apache.bookkeeper.proto.BookieClient; import org.apache.bookkeeper.util.OrderedSafeExecutor; import org.apache.log4j.Logger; - -import org.apache.zookeeper.Watcher; -import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooKeeper; import org.jboss.netty.channel.socket.ClientSocketChannelFactory; import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory; @@ -43,8 +42,8 @@ import org.jboss.netty.channel.socket.ni * BookKeeper client. We assume there is one single writer to a ledger at any * time. * - * There are three possible operations: start a new ledger, write to a ledger, - * and read from a ledger. + * There are four possible operations: start a new ledger, write to a ledger, + * read from a ledger and delete a ledger. * * The exceptions resulting from synchronous calls and error code resulting from * asynchronous calls can be found in the class {...@link BKException}. @@ -52,7 +51,7 @@ import org.jboss.netty.channel.socket.ni * */ -public class BookKeeper implements OpenCallback, CreateCallback { +public class BookKeeper implements OpenCallback, CreateCallback, DeleteCallback { static final Logger LOG = Logger.getLogger(BookKeeper.class); @@ -334,6 +333,56 @@ public class BookKeeper implements OpenC } /** + * Deletes a ledger asynchronously. + * + * @param lId + * ledger Id + * @param cb + * deleteCallback implementation + * @param ctx + * optional control object + */ + public void asyncDeleteLedger(long lId, DeleteCallback cb, Object ctx) { + new LedgerDeleteOp(this, lId, cb, ctx).initiate(); + } + + /** + * Delete callback implementation for synchronous delete call. + * + * @param rc + * return code + * @param ctx + * optional control object + */ + public void deleteComplete(int rc, Object ctx) { + SyncCounter counter = (SyncCounter) ctx; + counter.setrc(rc); + counter.dec(); + } + + /** + * Synchronous call to delete a ledger. Parameters match those of + * {...@link #asyncDeleteLedger(long, DeleteCallback, Object)} + * + * @param lId + * ledgerId + * @throws InterruptedException + * @throws BKException + */ + public void deleteLedger(long lId) throws InterruptedException, BKException { + SyncCounter counter = new SyncCounter(); + counter.inc(); + // Call asynchronous version + asyncDeleteLedger(lId, this, counter); + // Wait + counter.block(0); + if (counter.getrc() != KeeperException.Code.OK.intValue()) { + LOG.error("ZooKeeper error deleting ledger node: " + counter.getrc()); + throw BKException.create(Code.ZKException); + } + } + + /** * Shuts down client. * */ Modified: hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/CloseTest.java URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/CloseTest.java?rev=944003&r1=944002&r2=944003&view=diff ============================================================================== --- hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/CloseTest.java (original) +++ hadoop/zookeeper/trunk/src/contrib/bookkeeper/test/org/apache/bookkeeper/test/CloseTest.java Thu May 13 20:23:35 2010 @@ -33,7 +33,7 @@ import org.apache.log4j.Logger; */ public class CloseTest extends BaseTestCase{ - static Logger LOG = Logger.getLogger(LedgerRecoveryTest.class); + static Logger LOG = Logger.getLogger(CloseTest.class); DigestType digestType; public CloseTest(DigestType digestType) { Modified: hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperConfig.xml URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperConfig.xml?rev=944003&r1=944002&r2=944003&view=diff ============================================================================== --- hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperConfig.xml (original) +++ hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperConfig.xml Thu May 13 20:23:35 2010 @@ -100,8 +100,9 @@ </para> <para><computeroutput> - java -cp .:./zookeeper-<version>-bookkeeper.jar:./zookeeper-<version>.jar:../log4j/apache-log4j-1.2.15/log4j-1.2.15.jar\ - -Dlog4j.configuration=log4j.properties org.apache.bookkeeper.proto.BookieServer 3181 /path_to_log_device/\ + java -cp .:./zookeeper-<version>-bookkeeper.jar:./zookeeper-<version>.jar\ + :../log4j/apache-log4j-1.2.15/log4j-1.2.15.jar -Dlog4j.configuration=log4j.properties\ + org.apache.bookkeeper.proto.BookieServer 3181 127.0.0.1:2181 /path_to_log_device/\ /path_to_ledger_device/ </computeroutput></para> @@ -118,6 +119,12 @@ <listitem> <para> + Comma separated list of ZooKeeper servers with a hostname:port format; + </para> + </listitem> + + <listitem> + <para> Path for Log Device (stores bookie write-ahead log); </para> </listitem> @@ -146,4 +153,4 @@ </para> </section> </section> -</article> \ No newline at end of file +</article> Modified: hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperProgrammer.xml URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperProgrammer.xml?rev=944003&r1=944002&r2=944003&view=diff ============================================================================== --- hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperProgrammer.xml (original) +++ hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperProgrammer.xml Thu May 13 20:23:35 2010 @@ -69,6 +69,10 @@ <para><xref linkend="bk_readLedger" /></para> </listitem> + <listitem> + <para><xref linkend="bk_deleteLedger" /></para> + </listitem> + </itemizedlist> <section id="bk_instance"> @@ -604,5 +608,71 @@ </listitem> </itemizedlist> </section> + + <section id="bk_deleteLedger"> + <title> Deleting a ledger </title> + <para> + Once a client is done with a ledger and is sure that nobody will ever need to read from it again, they can delete the ledger. + The following methods belong to <computeroutput>org.apache.bookkeeper.client.BookKeeper</computeroutput>. + </para> + + <para> + <emphasis role="bold">Synchronous delete:</emphasis> + </para> + + <para> + <computeroutput> + public void deleteLedger(long lId) throws InterruptedException, BKException + </computeroutput> + </para> + + <itemizedlist> + <listitem> + <para> + <computeroutput>lId</computeroutput> is the ledger identifier; + </para> + </listitem> + </itemizedlist> + + <para> + <emphasis role="bold">Asynchronous delete:</emphasis> + </para> + <para> + <computeroutput> + public void asyncDeleteLedger(long lId, DeleteCallback cb, Object ctx) + </computeroutput> + </para> + + <para> + It takes a ledger identifier. Additionally, it takes a callback object + <computeroutput>cb</computeroutput> and a control object <computeroutput>ctx</computeroutput>. The callback object must implement + the <computeroutput>DeleteCallback</computeroutput> interface in <computeroutput>org.apache.bookkeeper.client.AsyncCallback</computeroutput>, and + a class implementing it has to implement a method called <computeroutput>deleteComplete</computeroutput> + that has the following signature: + </para> + + <para> + <computeroutput> + void deleteComplete(int rc, Object ctx) + </computeroutput> + </para> + + <para> + where: + </para> + <itemizedlist> + <listitem> + <para> + <computeroutput>rc</computeroutput> is a return code (please refer to <computeroutput>org.apache.bookeeper.client.BKDefs</computeroutput> for a list); + </para> + </listitem> + + <listitem> + <para> + <computeroutput>ctx</computeroutput> is control object used for accountability purposes. + </para> + </listitem> + </itemizedlist> + </section> </section> </article> \ No newline at end of file Modified: hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperStarted.xml URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperStarted.xml?rev=944003&r1=944002&r2=944003&view=diff ============================================================================== --- hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperStarted.xml (original) +++ hadoop/zookeeper/trunk/src/docs/src/documentation/content/xdocs/bookkeeperStarted.xml Thu May 13 20:23:35 2010 @@ -87,13 +87,17 @@ </para> <para><computeroutput> - java -cp .:./zookeeper-<version>-bookkeeper.jar:./zookeeper-<version>.jar:../log4j/apache-log4j-1.2.15/log4j-1.2.15.jar\ - -Dlog4j.configuration=log4j.properties org.apache.bookkeeper.proto.BookieServer 3181 /path_to_log_device/\ + java -cp .:./zookeeper-<version>-bookkeeper.jar:./zookeeper-<version>.jar\ + :../log4j/apache-log4j-1.2.15/log4j-1.2.15.jar -Dlog4j.configuration=log4j.properties\ + org.apache.bookkeeper.proto.BookieServer 3181 127.0.0.1:2181 /path_to_log_device/\ /path_to_ledger_device/ </computeroutput></para> <para> "/path_to_log_device/" and "/path_to_ledger_device/" are different paths. Also, port 3181 - is the port that a bookie listens on for connection requests from clients. + is the port that a bookie listens on for connection requests from clients. 127.0.0.1:2181 is the hostname:port + for the ZooKeeper server. In this example, the standalone ZooKeeper server is running locally on port 2181. + If we had multiple ZooKeeper servers, this parameter would be a comma separated list of all the hostname:port + values corresponding to them. </para> </section> @@ -201,4 +205,4 @@ lh.close(); </programlisting> </section> </section> -</article> \ No newline at end of file +</article>
