You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by th...@apache.org on 2014/02/12 16:25:48 UTC
svn commit: r1567646 - in /jackrabbit/oak/trunk/oak-mk: ./
src/main/java/org/apache/jackrabbit/mk/blobs/
src/test/java/org/apache/jackrabbit/mk/blobs/
Author: thomasm
Date: Wed Feb 12 15:25:48 2014
New Revision: 1567646
URL: http://svn.apache.org/r1567646
Log:
OAK-377 Data store garbage collection
Modified:
jackrabbit/oak/trunk/oak-mk/pom.xml
jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStore.java
jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/DbBlobStore.java
jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/FileBlobStore.java
jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/GarbageCollectableBlobStore.java
jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/MemoryBlobStore.java
jackrabbit/oak/trunk/oak-mk/src/test/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStoreTest.java
Modified: jackrabbit/oak/trunk/oak-mk/pom.xml
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-mk/pom.xml?rev=1567646&r1=1567645&r2=1567646&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-mk/pom.xml (original)
+++ jackrabbit/oak/trunk/oak-mk/pom.xml Wed Feb 12 15:25:48 2014
@@ -119,6 +119,13 @@
<scope>provided</scope>
</dependency>
+ <!-- apache commons dependencies -->
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>1.4</version>
+ </dependency>
+
<!--Test Dependencies-->
<dependency>
<groupId>com.googlecode.json-simple</groupId>
@@ -136,12 +143,6 @@
<artifactId>logback-classic</artifactId>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- <version>1.4</version>
- <scope>test</scope>
- </dependency>
</dependencies>
</project>
Modified: jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStore.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStore.java?rev=1567646&r1=1567645&r2=1567646&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStore.java (original)
+++ jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStore.java Wed Feb 12 15:25:48 2014
@@ -16,10 +16,6 @@
*/
package org.apache.jackrabbit.mk.blobs;
-import org.apache.jackrabbit.mk.util.Cache;
-import org.apache.jackrabbit.mk.util.IOUtils;
-import org.apache.jackrabbit.mk.util.StringUtils;
-
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
@@ -29,13 +25,20 @@ import java.io.InputStream;
import java.lang.ref.WeakReference;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Iterator;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.WeakHashMap;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.jackrabbit.mk.util.Cache;
+import org.apache.jackrabbit.mk.util.IOUtils;
+import org.apache.jackrabbit.mk.util.StringUtils;
+
/**
* An abstract data store that splits the binaries in relatively small blocks,
* so that each block fits in memory.
@@ -318,7 +321,7 @@ public abstract class AbstractBlobStore
}
}
- private byte[] readBlock(byte[] digest, long pos) {
+ byte[] readBlock(byte[] digest, long pos) {
BlockId id = new BlockId(digest, pos);
return cache.get(id).data;
}
@@ -421,6 +424,11 @@ public abstract class AbstractBlobStore
}
}
+ @Override
+ public Iterator<String> resolveChunks(String blobId) throws IOException {
+ return new ChunkIterator(blobId);
+ }
+
/**
* A block id. Blocks are small enough to fit in memory, so they can be
* cached.
@@ -493,4 +501,80 @@ public abstract class AbstractBlobStore
}
+ class ChunkIterator implements Iterator<String> {
+
+ private final static int BATCH = 2048;
+ private final ArrayDeque<String> queue;
+ private final ArrayDeque<ByteArrayInputStream> streamsStack;
+
+ public ChunkIterator(String blobId) {
+ byte[] id = StringUtils.convertHexToBytes(blobId);
+ ByteArrayInputStream idStream = new ByteArrayInputStream(id);
+ queue = new ArrayDeque<String>(BATCH);
+ streamsStack = new ArrayDeque<ByteArrayInputStream>();
+ streamsStack.push(idStream);
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (!queue.isEmpty()) {
+ return true;
+ }
+ try {
+ while ((queue.size() < BATCH)
+ && (streamsStack.peekFirst() != null)) {
+ ByteArrayInputStream idStream = streamsStack.peekFirst();
+
+ int type = idStream.read();
+ if (type == -1) {
+ streamsStack.pop();
+ } else if (type == TYPE_DATA) {
+ int len = IOUtils.readVarInt(idStream);
+ IOUtils.skipFully(idStream, len);
+ } else if (type == TYPE_HASH) {
+ int level = IOUtils.readVarInt(idStream);
+ // totalLength
+ IOUtils.readVarLong(idStream);
+ if (level > 0) {
+ // block length (ignored)
+ IOUtils.readVarLong(idStream);
+ }
+ byte[] digest = new byte[IOUtils.readVarInt(idStream)];
+ IOUtils.readFully(idStream, digest, 0, digest.length);
+ if (level > 0) {
+ queue.add(StringUtils.convertBytesToHex(digest));
+ byte[] block = readBlock(digest, 0);
+ idStream = new ByteArrayInputStream(block);
+ streamsStack.push(idStream);
+ } else {
+ queue.add(StringUtils.convertBytesToHex(digest));
+ }
+ } else {
+ break;
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ // Check now if ids available
+ if (!queue.isEmpty()) {
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ public String next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException("No data");
+ }
+ return queue.remove();
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException("Remove not supported");
+ }
+ }
}
Modified: jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/DbBlobStore.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/DbBlobStore.java?rev=1567646&r1=1567645&r2=1567646&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/DbBlobStore.java (original)
+++ jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/DbBlobStore.java Wed Feb 12 15:25:48 2014
@@ -16,9 +16,6 @@
*/
package org.apache.jackrabbit.mk.blobs;
-import org.apache.jackrabbit.mk.util.StringUtils;
-import org.h2.jdbcx.JdbcConnectionPool;
-
import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
@@ -26,6 +23,12 @@ import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
+import java.util.Iterator;
+
+import com.google.common.collect.AbstractIterator;
+
+import org.apache.jackrabbit.mk.util.StringUtils;
+import org.h2.jdbcx.JdbcConnectionPool;
/**
* A database blob store.
@@ -210,4 +213,68 @@ public class DbBlobStore extends Abstrac
return count;
}
+ @Override
+ public boolean deleteChunk(String chunkId) throws Exception {
+ Connection conn = cp.getConnection();
+ try {
+ PreparedStatement prep = conn
+ .prepareStatement("delete from datastore_meta where id = ?");
+ ;;;
+ // TODO and lastMod <= ?
+ ;;;
+ PreparedStatement prepData = conn
+ .prepareStatement("delete from datastore_data where id = ?");
+ prep.setString(1, chunkId);
+ prep.execute();
+ prepData.setString(1, chunkId);
+ prepData.execute();
+
+ prep.close();
+ prepData.close();
+ } finally {
+ conn.commit();
+ conn.close();
+ }
+
+ return true;
+ }
+
+ @Override
+ public Iterator<String> getAllChunkIds(long maxLastModifiedTime) throws Exception {
+ final Connection conn = cp.getConnection();
+ PreparedStatement prep = null;
+
+ if ((maxLastModifiedTime != 0) && (maxLastModifiedTime != -1)) {
+ prep = conn.prepareStatement(
+ "select id from datastore_meta where lastMod <= ?");
+ prep.setLong(1, maxLastModifiedTime);
+ } else {
+ prep = conn.prepareStatement(
+ "select id from datastore_meta");
+ }
+
+ final ResultSet rs = prep.executeQuery();
+
+ return new AbstractIterator<String>() {
+ @Override
+ protected String computeNext() {
+ try {
+ if (rs.next()) {
+ return rs.getString(1);
+ }
+ conn.close();
+ } catch (SQLException e) {
+ try {
+ if ((conn != null) && !conn.isClosed()) {
+ conn.close();
+ }
+ } catch (Exception e2) {
+ // ignore
+ }
+ throw new RuntimeException(e);
+ }
+ return endOfData();
+ }
+ };
+ }
}
Modified: jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/FileBlobStore.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/FileBlobStore.java?rev=1567646&r1=1567645&r2=1567646&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/FileBlobStore.java (original)
+++ jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/FileBlobStore.java Wed Feb 12 15:25:48 2014
@@ -16,9 +16,6 @@
*/
package org.apache.jackrabbit.mk.blobs;
-import org.apache.jackrabbit.mk.util.IOUtils;
-import org.apache.jackrabbit.mk.util.StringUtils;
-
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
@@ -29,6 +26,19 @@ import java.io.OutputStream;
import java.security.DigestInputStream;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
+import java.util.Iterator;
+
+import javax.annotation.Nullable;
+
+import com.google.common.base.Predicate;
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.FluentIterable;
+import com.google.common.io.Files;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.io.FilenameUtils;
+import org.apache.jackrabbit.mk.util.IOUtils;
+import org.apache.jackrabbit.mk.util.StringUtils;
/**
* A file blob store.
@@ -36,6 +46,7 @@ import java.security.NoSuchAlgorithmExce
public class FileBlobStore extends AbstractBlobStore {
private static final String OLD_SUFFIX = "_old";
+ private static final String FILE_SUFFIX = ".dat";
private final File baseDir;
private final byte[] buffer = new byte[16 * 1024];
@@ -117,7 +128,7 @@ public class FileBlobStore extends Abstr
if (old) {
sub2 += OLD_SUFFIX;
}
- return new File(new File(new File(baseDir, sub1), sub2), id + ".dat");
+ return new File(new File(new File(baseDir, sub1), sub2), id + FILE_SUFFIX);
}
@Override
@@ -207,4 +218,45 @@ public class FileBlobStore extends Abstr
return count;
}
+ @Override
+ public boolean deleteChunk(String chunkId) throws Exception {
+ byte[] digest = StringUtils.convertHexToBytes(chunkId);
+ File f = getFile(digest, false);
+ if (!f.exists()) {
+ File old = getFile(digest, true);
+ f.getParentFile().mkdir();
+ old.renameTo(f);
+ f = getFile(digest, false);
+ }
+ f.delete();
+ return mark;
+ }
+
+ @Override
+ public Iterator<String> getAllChunkIds(final long maxLastModifiedTime) throws Exception {
+ FluentIterable<File> iterable = Files.fileTreeTraverser().postOrderTraversal(baseDir);
+ final Iterator<File> iter =
+ iterable.filter(new Predicate<File>() {
+ // Ignore the directories and files newer than maxLastModifiedTime if specified
+ @Override
+ public boolean apply(@Nullable File input) {
+ if (!input.isDirectory() && (
+ (maxLastModifiedTime == 0 || maxLastModifiedTime == -1) ||
+ FileUtils.isFileOlder(input, maxLastModifiedTime))) {
+ return true;
+ }
+ return false;
+ }
+ }).iterator();
+ return new AbstractIterator<String>() {
+ @Override
+ protected String computeNext() {
+ if (iter.hasNext()) {
+ File file = iter.next();
+ return FilenameUtils.removeExtension(file.getName());
+ }
+ return endOfData();
+ }
+ };
+ }
}
\ No newline at end of file
Modified: jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/GarbageCollectableBlobStore.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/GarbageCollectableBlobStore.java?rev=1567646&r1=1567645&r2=1567646&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/GarbageCollectableBlobStore.java (original)
+++ jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/GarbageCollectableBlobStore.java Wed Feb 12 15:25:48 2014
@@ -17,6 +17,7 @@
package org.apache.jackrabbit.mk.blobs;
import java.io.IOException;
+import java.util.Iterator;
/**
* A blob store that support garbage collection.
@@ -70,4 +71,34 @@ public interface GarbageCollectableBlobS
*/
long getBlockSizeMin();
+ /**
+ * Gets all the identifiers.
+ *
+ * @param maxLastModifiedTime
+ * the max last modified time to consider for retrieval
+ * @return the identifiers
+ * @throws Exception
+ * the exception
+ */
+ Iterator<String> getAllChunkIds(long maxLastModifiedTime) throws Exception;
+
+ /**
+ * Delete the blob with the given id.
+ *
+ * @param chunkId the chunk id
+ * @return true, if successful
+ * @throws Exception
+ * the exception
+ */
+ boolean deleteChunk(String chunkId) throws Exception;
+
+ /**
+ * Resolve chunks from the given Id.
+ *
+ * @param blobId the blob id
+ * @return the iterator
+ * @throws IOException
+ * Signals that an I/O exception has occurred.
+ */
+ Iterator<String> resolveChunks(String blobId) throws IOException;
}
Modified: jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/MemoryBlobStore.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/MemoryBlobStore.java?rev=1567646&r1=1567645&r2=1567646&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/MemoryBlobStore.java (original)
+++ jackrabbit/oak/trunk/oak-mk/src/main/java/org/apache/jackrabbit/mk/blobs/MemoryBlobStore.java Wed Feb 12 15:25:48 2014
@@ -18,6 +18,12 @@ package org.apache.jackrabbit.mk.blobs;
import java.io.IOException;
import java.util.HashMap;
+import java.util.Iterator;
+
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.Maps;
+
+import org.apache.jackrabbit.mk.util.StringUtils;
/**
* A memory blob store. Useful for testing.
@@ -74,4 +80,39 @@ public class MemoryBlobStore extends Abs
return count;
}
+ @Override
+ public boolean deleteChunk(String chunkId) throws Exception {
+ BlockId id = new BlockId(StringUtils.convertHexToBytes(chunkId), 0);
+ if (map.containsKey(id)) {
+ map.remove(id);
+ } else if (old.containsKey(id)) {
+ old.remove(id);
+ }
+ return true;
+ }
+
+ /**
+ * Ignores the maxlastModifiedTime
+ */
+ @Override
+ public Iterator<String> getAllChunkIds(long maxLastModifiedTime) throws Exception {
+ HashMap<BlockId, byte[]> combinedMap = Maps.newHashMap();
+ combinedMap.putAll(map);
+ combinedMap.putAll(old);
+ final Iterator<BlockId> iter = combinedMap.keySet().iterator();
+
+ return new AbstractIterator<String>() {
+ @Override
+ protected String computeNext() {
+ if (iter.hasNext()) {
+ BlockId blockId = iter.next();
+ if (blockId != null) {
+ return StringUtils.convertBytesToHex(blockId.getDigest());
+ }
+ }
+ return endOfData();
+ }
+ };
+ }
+
}
Modified: jackrabbit/oak/trunk/oak-mk/src/test/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStoreTest.java
URL: http://svn.apache.org/viewvc/jackrabbit/oak/trunk/oak-mk/src/test/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStoreTest.java?rev=1567646&r1=1567645&r2=1567646&view=diff
==============================================================================
--- jackrabbit/oak/trunk/oak-mk/src/test/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStoreTest.java (original)
+++ jackrabbit/oak/trunk/oak-mk/src/test/java/org/apache/jackrabbit/mk/blobs/AbstractBlobStoreTest.java Wed Feb 12 15:25:48 2014
@@ -26,13 +26,19 @@ import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.Random;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import junit.framework.TestCase;
+
+import com.google.common.collect.Sets;
+
import org.apache.jackrabbit.mk.json.JsopBuilder;
import org.apache.jackrabbit.mk.json.JsopTokenizer;
import org.apache.jackrabbit.mk.util.IOUtilsTest;
+import org.junit.Test;
/**
* Tests a BlobStore implementation.
@@ -44,8 +50,10 @@ public abstract class AbstractBlobStoreT
/**
* Should be overridden by subclasses to set the {@link #store} variable.
*/
+ @Override
public abstract void setUp() throws Exception;
+ @Override
public void tearDown() throws Exception {
store = null;
}
@@ -93,9 +101,11 @@ public abstract class AbstractBlobStoreT
public void testCloseStream() throws Exception {
final AtomicBoolean closed = new AtomicBoolean();
InputStream in = new InputStream() {
+ @Override
public void close() {
closed.set(true);
}
+ @Override
public int read() throws IOException {
return -1;
}
@@ -107,9 +117,11 @@ public abstract class AbstractBlobStoreT
public void testExceptionWhileReading() throws Exception {
final AtomicBoolean closed = new AtomicBoolean();
InputStream in = new InputStream() {
+ @Override
public void close() {
closed.set(true);
}
+ @Override
public int read() throws IOException {
throw new RuntimeException("abc");
}
@@ -335,4 +347,52 @@ public abstract class AbstractBlobStoreT
list.add(file.getAbsolutePath());
}
+ @Test
+ public void testList() throws Exception {
+ Set<String> ids = createArtifacts();
+
+ Iterator<String> iter = store.getAllChunkIds(0);
+ while (iter.hasNext()) {
+ ids.remove(iter.next());
+ }
+
+ assertTrue(ids.isEmpty());
+ }
+
+ @Test
+ public void testDelete() throws Exception {
+ Set<String> ids = createArtifacts();
+
+ for (String id : ids) {
+ store.deleteChunk(id);
+ }
+
+ Iterator<String> iter = store.getAllChunkIds(0);
+ Set<String> ret = Sets.newHashSet();
+ while (iter.hasNext()) {
+ ret.add(iter.next());
+ }
+
+ assertTrue(ret.isEmpty());
+ }
+
+ private Set<String> createArtifacts() throws Exception {
+ Set<String> ids = Sets.newHashSet();
+ int number = 10;
+ for (int i = 0; i < number; i++) {
+ String id = store.writeBlob(randomStream(i, 4160));
+ Iterator<String> iter = store.resolveChunks(id.toString());
+ while (iter.hasNext()) {
+ ids.add(iter.next());
+ }
+ }
+ return ids;
+ }
+
+ static InputStream randomStream(int seed, int size) {
+ Random r = new Random(seed);
+ byte[] data = new byte[size];
+ r.nextBytes(data);
+ return new ByteArrayInputStream(data);
+ }
}