Author: adriancole Date: 2009-03-08 17:03:03 -0400 (Sun, 08 Mar 2009) New Revision: 7884 Added: core/branches/flat/src/main/java/org/horizon/loader/jdbm/ core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmCacheStore.java core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmCacheStoreConfig.java core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmConnection.java core/branches/flat/src/test/java/org/horizon/loader/jdbm/ core/branches/flat/src/test/java/org/horizon/loader/jdbm/JdbmCacheStoreTest.java core/branches/flat/src/test/java/org/horizon/loader/jdbm/JdbmLearningTest.java Modified: core/branches/flat/pom.xml Log: added jdbm cachestore Modified: core/branches/flat/pom.xml =================================================================== --- core/branches/flat/pom.xml 2009-03-08 21:02:07 UTC (rev 7883) +++ core/branches/flat/pom.xml 2009-03-08 21:03:03 UTC (rev 7884) @@ -52,6 +52,13 @@ <version>1.0</version> <optional>true</optional> </dependency> + <!-- needed for jdbm --> + <dependency> + <groupId>commons-collections</groupId> + <artifactId>commons-collections</artifactId> + <version>3.2.1</version> + <optional>true</optional> + </dependency> <dependency> <groupId>c3p0</groupId> Copied: core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmCacheStore.java (from rev 7868, core/branches/flat/src/main/java/org/horizon/loader/file/FileCacheStore.java) =================================================================== --- core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmCacheStore.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmCacheStore.java 2009-03-08 21:03:03 UTC (rev 7884) @@ -0,0 +1,166 @@ +package org.horizon.loader.jdbm; + +import jdbm.btree.BTree; +import jdbm.helper.Tuple; +import jdbm.helper.TupleBrowser; +import org.horizon.Cache; +import org.horizon.loader.CacheLoaderConfig; +import org.horizon.loader.CacheLoaderException; +import org.horizon.loader.StoredEntry; +import org.horizon.loader.bucket.Bucket; +import org.horizon.loader.file.FileCacheStore; +import org.horizon.marshall.Marshaller; + +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.HashSet; +import java.util.Set; + +/** + * A JDBM (http://jdbm.sourceforge.net) extension of {@link org.horizon.loader.bucket.BucketBasedCacheStore}. This file + * store stores stuff in a {@link BTree} at the following location: <tt>/{location}/cache name/db.db</tt>. The {@link + * BTree} is named the same as {@link org.horizon.Cache#getName()}. {@link Bucket}s are stored inside the {@link BTree} + * keyed on {@link org.horizon.loader.bucket.Bucket#getBucketName()}. + * <p/> + * <p/> + * JdbmCacheStore uses {@link Bucket}s as entries in {@Btree} are limited by having <tt>int</tt> primary keys. + * <p/> + * JdbmCacheStore does not use JDBM transactions and instead relies on the locking mechanisms provided by {@link + * org.horizon.loader.LockSupportCacheStore}. + * + * @author Adrian Cole + * @since 1.0 + */ +public class JdbmCacheStore extends FileCacheStore { + + private BTree btree; + private String treeName; + private JdbmConnection connection; + + @Override + public void init(CacheLoaderConfig config, Cache cache, Marshaller m) { + init((JdbmCacheStoreConfig) config, cache, m, new JdbmConnection()); + } + + public void init(JdbmCacheStoreConfig config, Cache cache, Marshaller m, JdbmConnection connection) { + super.init(config, cache, m); + this.treeName = cache.getName(); + this.connection = connection; + } + + public Class<? extends CacheLoaderConfig> getConfigurationClass() { + return JdbmCacheStoreConfig.class; + } + + @Override + public void start() throws CacheLoaderException { + super.start(); + open(); + } + + private void open() throws CacheLoaderException { + connection.openEnvironment(getRoot()); + try { + btree = connection.loadOrCreateBTree(treeName); + } catch (IOException caught) { + throw new CacheLoaderException("Problem loading btree", caught); + } + } + + @Override + public void stop() throws CacheLoaderException { + super.stop(); + close(); + } + + private void close() throws CacheLoaderException { + btree = null; + connection.closeEnvironment(); + } + + @Override + protected Set<StoredEntry> loadAllLockSafe() throws CacheLoaderException { + Set<StoredEntry> result = new HashSet<StoredEntry>(); + try { + TupleBrowser browser = btree.browse(); + Tuple tuple = new Tuple(); + while (browser.getNext(tuple)) { + if (tuple.getValue() != null) { + Bucket bucket = (Bucket) tuple.getValue(); + if (bucket != null) { + if (bucket.removeExpiredEntries()) { + saveBucket(bucket); + } + result.addAll(bucket.getStoredEntries()); + } + } + } + } catch (IOException e) { + throw new CacheLoaderException("Problem loading entries", e); + } + return result; + } + + @Override + protected void fromStreamLockSafe(ObjectInput objectInput) throws CacheLoaderException { + try { + close(); + String name = (String) objectInput.readObject(); + super.fromStreamLockSafe(objectInput); + String oldName = treeName; + treeName = name; + open(); + connection.associateBTreeWithName(btree, oldName); + treeName = oldName; + } catch (Exception e) { + throw new CacheLoaderException("Problem loading from stream", e); + } + } + + @Override + protected void toStreamLockSafe(ObjectOutput objectOutput) throws CacheLoaderException { + try { + close(); + objectOutput.writeObject(treeName); + super.toStreamLockSafe(objectOutput); + } catch (IOException e) { + throw new CacheLoaderException("Problem generating stream", e); + } finally { + open(); + } + } + + @Override + protected void clearLockSafe() throws CacheLoaderException { + try { + connection.deleteBTree(btree); + btree = connection.createBTree(treeName); + } catch (IOException caught) { + throw convertToCacheLoaderException("error recreating tree" + treeName, caught); + } + } + + @Override + protected Bucket loadBucket(String bucketName) throws CacheLoaderException { + try { + return (Bucket) btree.find(bucketName); + } catch (IOException caught) { + throw convertToCacheLoaderException("error finding bucket " + bucketName, caught); + } + } + + @Override + public void saveBucket(Bucket b) throws CacheLoaderException { + try { + btree.insert(b.getBucketName(), b, true); + } catch (IOException caught) { + throw convertToCacheLoaderException("error saving bucket " + b.getBucketName(), caught); + } + } + + CacheLoaderException convertToCacheLoaderException(String message, Exception caught) { + return (caught instanceof CacheLoaderException) ? (CacheLoaderException) caught : + new CacheLoaderException(message, caught); + } +} \ No newline at end of file Copied: core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmCacheStoreConfig.java (from rev 7868, core/branches/flat/src/main/java/org/horizon/loader/file/FileCacheStoreConfig.java) =================================================================== --- core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmCacheStoreConfig.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmCacheStoreConfig.java 2009-03-08 21:03:03 UTC (rev 7884) @@ -0,0 +1,19 @@ +package org.horizon.loader.jdbm; + +import org.horizon.loader.file.FileCacheStoreConfig; + +/** + * Configures {@link org.horizon.loader.jdbm.JdbmCacheStore}. + * + * @author Adrian Cole + * @see FileCacheStoreConfig + * @since 1.0 + */ +public class JdbmCacheStoreConfig extends FileCacheStoreConfig { + + public JdbmCacheStoreConfig() { + super(); + setLocation("Horizon-JdbmCacheStore"); + setCacheLoaderClassName(JdbmCacheStore.class.getName()); + } +} \ No newline at end of file Added: core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmConnection.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmConnection.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/loader/jdbm/JdbmConnection.java 2009-03-08 21:03:03 UTC (rev 7884) @@ -0,0 +1,81 @@ +package org.horizon.loader.jdbm; + +import jdbm.RecordManager; +import jdbm.RecordManagerFactory; +import jdbm.btree.BTree; +import org.apache.commons.collections.ComparatorUtils; +import org.horizon.loader.CacheLoaderException; + +import java.io.File; +import java.io.IOException; +import java.util.Comparator; +import java.util.Properties; + +/** + * This interface defines the interactons between the {@link org.horizon.loader.jdbm.JdbmCacheStore} and the JDBM {@link + * RecordManager}. + * + * @author Adrian Cole + * @since 1.0 + */ +public class JdbmConnection { + + private RecordManager recman; + + public void openEnvironment(File root) throws CacheLoaderException { + System.out.println("Opening environment in: " + root); + try { + recman = createRecordManager(root); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + + public void closeEnvironment() throws CacheLoaderException { + try { + recman.close(); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + recman = null; + } + + + public BTree createBTree(String name) throws IOException { + Comparator comp = ComparatorUtils.naturalComparator(); + /* note the code of btree does not expose its serializer and key/value serializers + that are exposed are never used. Accordingly, we don't set the serializer */ + BTree tree = BTree.createInstance(recman, comp); + associateBTreeWithName(tree, name); + return tree; + } + + public void associateBTreeWithName(BTree tree, String name) throws IOException { + recman.setNamedObject(name, tree.getRecid()); + } + + public RecordManager createRecordManager(File home) throws IOException { + return RecordManagerFactory.createRecordManager(new File(home, "/db").getAbsolutePath(), new Properties()); + } + + public BTree loadOrCreateBTree(String name) throws IOException { + BTree btree; + try { + btree = loadBTree(name); + } catch (IllegalArgumentException e) { + btree = createBTree(name); + } + return btree; + } + + public void deleteBTree(BTree btree) throws IOException { + recman.delete(btree.getRecid()); + } + + public BTree loadBTree(String name) throws IOException { + long recid = recman.getNamedObject(name); + if (recid != 0) + return BTree.load(recman, recid); + else throw new IllegalArgumentException("name not found in record manager" + name); + } +} \ No newline at end of file Copied: core/branches/flat/src/test/java/org/horizon/loader/jdbm/JdbmCacheStoreTest.java (from rev 7868, core/branches/flat/src/test/java/org/horizon/loader/file/FileCacheStoreTest.java) =================================================================== --- core/branches/flat/src/test/java/org/horizon/loader/jdbm/JdbmCacheStoreTest.java (rev 0) +++ core/branches/flat/src/test/java/org/horizon/loader/jdbm/JdbmCacheStoreTest.java 2009-03-08 21:03:03 UTC (rev 7884) @@ -0,0 +1,36 @@ +package org.horizon.loader.jdbm; + +import org.horizon.loader.BaseCacheStoreTest; +import org.horizon.loader.CacheLoaderException; +import org.horizon.loader.CacheStore; +import org.horizon.test.TestingUtil; +import org.testng.annotations.AfterTest; +import org.testng.annotations.BeforeTest; +import org.testng.annotations.Test; + +import java.io.File; + +@Test(groups = "unit", testName = "loader.jdbm.JdbmCacheStoreTest") +public class JdbmCacheStoreTest extends BaseCacheStoreTest { + + private final String tmpDirectory = TestingUtil.TEST_FILES + File.separator + getClass().getSimpleName(); + private JdbmCacheStore jcs; + + protected CacheStore createCacheStore() throws CacheLoaderException { + jcs = new JdbmCacheStore(); + JdbmCacheStoreConfig cfg = new JdbmCacheStoreConfig(); + cfg.setLocation(tmpDirectory); + cfg.setPurgeSynchronously(true); // for more accurate unit testing + jcs.init(cfg, getCache(), getMarshaller()); + jcs.start(); + return jcs; + } + + @AfterTest + @BeforeTest + public void removeTempDirectory() { + TestingUtil.recursiveFileRemove(tmpDirectory); + } + + +} \ No newline at end of file Added: core/branches/flat/src/test/java/org/horizon/loader/jdbm/JdbmLearningTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/loader/jdbm/JdbmLearningTest.java (rev 0) +++ core/branches/flat/src/test/java/org/horizon/loader/jdbm/JdbmLearningTest.java 2009-03-08 21:03:03 UTC (rev 7884) @@ -0,0 +1,764 @@ +package org.horizon.loader.jdbm; + +import jdbm.RecordManager; +import jdbm.RecordManagerFactory; +import jdbm.btree.BTree; +import jdbm.helper.Tuple; +import jdbm.helper.TupleBrowser; +import org.apache.commons.collections.ComparatorUtils; +import org.easymock.EasyMock; +import org.horizon.io.UnclosableObjectInputStream; +import org.horizon.io.UnclosableObjectOutputStream; +import org.horizon.loader.CacheLoaderException; +import org.horizon.loader.StoredEntry; +import org.horizon.loader.modifications.Clear; +import org.horizon.loader.modifications.Modification; +import org.horizon.loader.modifications.Remove; +import org.horizon.loader.modifications.Store; +import org.horizon.logging.Log; +import org.horizon.logging.LogFactory; +import org.horizon.test.TestingUtil; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.BeforeTest; +import org.testng.annotations.Test; + +import javax.transaction.Transaction; +import java.io.*; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Learning tests for JDBM (http://jdbm.sourceforge.net). Behaviour here is used in JdbmCacheLoader. When there are + * upgrades to JDBM, this test may warrant updating. + * + * @author Adrian Cole + * @version $Id: $ + * @since 1.0 + */ +@Test(groups = "unit", enabled = true, testName = "loader.bdbje.JdbmLearningTest") +public class JdbmLearningTest { + + private static final Log log = LogFactory.getLog(JdbmLearningTest.class); + private static final boolean trace = log.isTraceEnabled(); + + String dbHome = TestingUtil.TEST_FILES + "/Horizon-JdbmLearningTest"; + private BTree btree; + private RecordManager recman; + private String treeName = this.getClass().getName(); + + @BeforeTest + @AfterMethod + public void tearDown() throws Exception { + TestingUtil.recursiveFileRemove(dbHome); + } + + + @BeforeMethod + public void setUp() throws CacheLoaderException { + new File(dbHome).mkdirs(); + start(); + + } + + private BTree createBTree(RecordManager recman, String name) throws IOException { + + Comparator comp = ComparatorUtils.naturalComparator(); + /* note the code of btree does not expose its serializer and key/value serializers + that are exposed are never used. Accordingly, we don't set the serializer */ + BTree tree = BTree.createInstance(recman, comp); + associateBTreeWithName(recman, tree, name); + return tree; + } + + private void associateBTreeWithName(RecordManager recman, BTree tree, String name) throws IOException { + recman.setNamedObject(name, tree.getRecid()); + } + + private RecordManager createRecordManager(String home) throws IOException { + return RecordManagerFactory.createRecordManager(home + "/db", new Properties()); + } + + private void start() throws CacheLoaderException { + System.out.println("Opening environment in: " + dbHome); + try { + recman = createRecordManager(dbHome); + btree = loadOrCreateBTree(recman, treeName); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + + private BTree loadOrCreateBTree(RecordManager recman, String name) throws IOException { + BTree btree; + try { + btree = loadBTree(recman, name); + } catch (IllegalArgumentException e) { + btree = createBTree(recman, name); + } + return btree; + } + + private void deleteBTree(RecordManager recman, BTree btree) throws IOException { + recman.delete(btree.getRecid()); + } + + private BTree loadBTree(RecordManager recman, String name) throws IOException { + long recid = recman.getNamedObject(name); + log.debug(name + " located as " + recid); + if (recid != 0) + return BTree.load(recman, recid); + else throw new IllegalArgumentException("name not found in record manager" + name); + } + + private void stop() throws CacheLoaderException { + btree = null; + try { + recman.close(); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + recman = null; + } + + + public boolean containsKey(Object key) throws CacheLoaderException { + return load(key) != null; + } + + public StoredEntry load(Object key) throws CacheLoaderException { + try { + StoredEntry entry = (StoredEntry) btree.find(key); + return removeAndReturnNullIfExpired(entry); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + + private StoredEntry removeAndReturnNullIfExpired(StoredEntry entry) throws IOException { + if (entry != null && entry.isExpired()) { + btree.remove(entry.getKey()); + return null; + } + return entry; + } + + public Set<StoredEntry> loadAll() throws CacheLoaderException { + try { + Set<StoredEntry> entries = new HashSet<StoredEntry>(); + TupleBrowser browser = btree.browse(); + Tuple tuple = new Tuple(); + while (browser.getNext(tuple)) { + if (tuple.getValue() != null) { + StoredEntry entry = (StoredEntry) tuple.getValue(); + entry = removeAndReturnNullIfExpired(entry); + if (entry != null) entries.add(entry); + } + } + return entries; + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + + public void store(StoredEntry ed) throws CacheLoaderException { + try { + btree.insert(ed.getKey(), ed, true); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + + private void purgeExpired() throws CacheLoaderException { + try { + TupleBrowser browser = btree.browse(); + Tuple tuple = new Tuple(); + while (browser.getNext(tuple)) { + if (tuple.getValue() != null) { + StoredEntry entry = (StoredEntry) tuple.getValue(); + removeAndReturnNullIfExpired(entry); + } + } + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + + + public void fromStream(ObjectInput inputStream) throws CacheLoaderException { + FileChannel channel = null; + try { + stop(); + String name = (String) inputStream.readObject(); + + channel = new RandomAccessFile(dbHome + "/db.db", "rw").getChannel(); + long size = inputStream.readLong(); + ByteBuffer buff = channel.map(FileChannel.MapMode.READ_WRITE, 0, size); + long i = 0; + while (i++ < size) { + buff.put(inputStream.readByte()); + } + channel.close(); + + channel = new RandomAccessFile(dbHome + "/db.lg", "rw").getChannel(); + size = inputStream.readLong(); + buff = channel.map(FileChannel.MapMode.READ_WRITE, 0, size); + i = 0; + while (i++ < size) { + buff.put(inputStream.readByte()); + } + channel.close(); + + String oldName = treeName; + treeName = name; + start(); + associateBTreeWithName(recman, btree, oldName); + treeName = oldName; + + } catch (FileNotFoundException e) { + throw new CacheLoaderException(e); + } catch (IOException e) { + throw new CacheLoaderException(e); + } catch (ClassNotFoundException e) { + throw new CacheLoaderException(e); + } finally { + if (channel != null) + try { + channel.close(); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + } + + public void toStream(ObjectOutput outputStream) throws CacheLoaderException { + FileChannel channel = null; + try { + stop(); + outputStream.writeObject(treeName); + + channel = new RandomAccessFile(dbHome + "/db.db", "r").getChannel(); + long size = channel.size(); + outputStream.writeLong(size); + ByteBuffer buff = channel.map(FileChannel.MapMode.READ_ONLY, 0, size); + long i = 0; + while (i++ < size) { + outputStream.write(buff.get()); + } + channel.close(); + + channel = new RandomAccessFile(dbHome + "/db.lg", "r").getChannel(); + size = channel.size(); + outputStream.writeLong(size); + buff = channel.map(FileChannel.MapMode.READ_ONLY, 0, size); + i = 0; + while (i++ < size) { + outputStream.write(buff.get()); + } + channel.close(); + channel = null; + + } catch (FileNotFoundException e) { + throw new CacheLoaderException(e); + } catch (IOException e) { + throw new CacheLoaderException(e); + } finally { + if (recman == null) + start(); + if (channel != null) + try { + channel.close(); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + } + + public void clear() throws CacheLoaderException { + try { + deleteBTree(recman, btree); + btree = createBTree(recman, treeName); + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + + public boolean remove(Object key) throws CacheLoaderException { + try { + return btree.remove(key) != null; + } catch (IllegalArgumentException e) { + return false; + } catch (IOException e) { + throw new CacheLoaderException(e); + } + } + + + private final Map<Transaction, List<? extends Modification>> transactions = new ConcurrentHashMap<Transaction, List<? extends Modification>>(); + + protected void applyModifications(List<? extends Modification> mods) throws CacheLoaderException { + for (Modification m : mods) { + switch (m.getType()) { + case STORE: + Store s = (Store) m; + store(s.getStoredEntry()); + break; + case CLEAR: + clear(); + break; + case REMOVE: + Remove r = (Remove) m; + remove(r.getKey()); + break; + default: + throw new IllegalArgumentException("Unknown modification type " + m.getType()); + } + } + } + + public void prepare(List<? extends Modification> mods, Transaction tx, boolean isOnePhase) throws CacheLoaderException { + if (isOnePhase) { + applyModifications(mods); + } else { + transactions.put(tx, mods); + } + } + + public void rollback(Transaction tx) { + transactions.remove(tx); + } + + public void commit(Transaction tx) throws CacheLoaderException { + List<? extends Modification> list = transactions.remove(tx); + if (list != null && !list.isEmpty()) applyModifications(list); + } + + public void removeAll(Set<Object> keys) throws CacheLoaderException { + if (keys != null && !keys.isEmpty()) { + for (Object key : keys) remove(key); + } + } + + + public void testLoadAndStore() throws InterruptedException, CacheLoaderException { + assert !containsKey("k"); + StoredEntry se = new StoredEntry("k", "v", -1, -1); + store(se); + + assert load("k").getValue().equals("v"); + assert load("k").getLifespan() == -1; + assert !load("k").isExpired(); + assert containsKey("k"); + + long now = System.currentTimeMillis(); + long lifespan = 120000; + se = new StoredEntry("k", "v", now, now + lifespan); + store(se); + + assert load("k").getValue().equals("v"); + assert load("k").getLifespan() == lifespan; + assert !load("k").isExpired(); + assert containsKey("k"); + + now = System.currentTimeMillis(); + lifespan = 1; + se = new StoredEntry("k", "v", now, now + lifespan); + store(se); + Thread.sleep(100); + assert se.isExpired(); + assert load("k") == null; + assert !containsKey("k"); + } + + + public void testStopStartDoesntNukeValues() throws InterruptedException, CacheLoaderException { + assert !containsKey("k1"); + assert !containsKey("k2"); + + long now = System.currentTimeMillis(); + long lifespan = 1; + StoredEntry se1 = new StoredEntry("k1", "v1", now, now + lifespan); + StoredEntry se2 = new StoredEntry("k2", "v2"); + + store(se1); + store(se2); + Thread.sleep(100); + stop(); + start(); + assert se1.isExpired(); + assert load("k1") == null; + assert !containsKey("k1"); + assert load("k2") != null; + assert containsKey("k2"); + assert load("k2").getValue().equals("v2"); + + } + + + public void testOnePhaseCommit() throws CacheLoaderException { + List<Modification> mods = new ArrayList<Modification>(); + mods.add(new Store(new StoredEntry("k1", "v1", -1, -1))); + mods.add(new Store(new StoredEntry("k2", "v2", -1, -1))); + mods.add(new Remove("k1")); + Transaction tx = EasyMock.createNiceMock(Transaction.class); + prepare(mods, tx, true); + + assert load("k2").getValue().equals("v2"); + assert !containsKey("k1"); + + clear(); + + mods = new ArrayList<Modification>(); + mods.add(new Store(new StoredEntry("k1", "v1", -1, -1))); + mods.add(new Store(new StoredEntry("k2", "v2", -1, -1))); + mods.add(new Clear()); + mods.add(new Store(new StoredEntry("k3", "v3", -1, -1))); + + prepare(mods, tx, true); + assert !containsKey("k1"); + assert !containsKey("k2"); + assert containsKey("k3"); + } + + public void testTwoPhaseCommit() throws CacheLoaderException { + List<Modification> mods = new ArrayList<Modification>(); + mods.add(new Store(new StoredEntry("k1", "v1", -1, -1))); + mods.add(new Store(new StoredEntry("k2", "v2", -1, -1))); + mods.add(new Remove("k1")); + Transaction tx = EasyMock.createNiceMock(Transaction.class); + prepare(mods, tx, false); + + assert !containsKey("k1"); + assert !containsKey("k2"); + + commit(tx); + + assert load("k2").getValue().equals("v2"); + assert !containsKey("k1"); + + clear(); + + mods = new ArrayList<Modification>(); + mods.add(new Store(new StoredEntry("k1", "v1", -1, -1))); + mods.add(new Store(new StoredEntry("k2", "v2", -1, -1))); + mods.add(new Clear()); + mods.add(new Store(new StoredEntry("k3", "v3", -1, -1))); + + prepare(mods, tx, false); + + assert !containsKey("k1"); + assert !containsKey("k2"); + assert !containsKey("k3"); + + commit(tx); + + assert !containsKey("k1"); + assert !containsKey("k2"); + assert containsKey("k3"); + } + + + public void testRollback() throws CacheLoaderException { + + store(new StoredEntry("old", "old", -1, -1)); + + List<Modification> mods = new ArrayList<Modification>(); + mods.add(new Store(new StoredEntry("k1", "v1", -1, -1))); + mods.add(new Store(new StoredEntry("k2", "v2", -1, -1))); + mods.add(new Remove("k1")); + mods.add(new Remove("old")); + Transaction tx = EasyMock.createNiceMock(Transaction.class); + prepare(mods, tx, false); + + assert !containsKey("k1"); + assert !containsKey("k2"); + assert containsKey("old"); + + rollback(tx); + + assert !containsKey("k1"); + assert !containsKey("k2"); + assert containsKey("old"); + + mods = new ArrayList<Modification>(); + mods.add(new Store(new StoredEntry("k1", "v1", -1, -1))); + mods.add(new Store(new StoredEntry("k2", "v2", -1, -1))); + mods.add(new Clear()); + mods.add(new Store(new StoredEntry("k3", "v3", -1, -1))); + + prepare(mods, tx, false); + + assert !containsKey("k1"); + assert !containsKey("k2"); + assert !containsKey("k3"); + + rollback(tx); + + assert !containsKey("k1"); + assert !containsKey("k2"); + assert !containsKey("k3"); + assert containsKey("old"); + } + + public void testRollbackFromADifferentThreadReusingTransactionKey() throws CacheLoaderException, InterruptedException { + + store(new StoredEntry("old", "old", -1, -1)); + + List<Modification> mods = new ArrayList<Modification>(); + mods.add(new Store(new StoredEntry("k1", "v1", -1, -1))); + mods.add(new Store(new StoredEntry("k2", "v2", -1, -1))); + mods.add(new Remove("k1")); + mods.add(new Remove("old")); + final Transaction tx = EasyMock.createNiceMock(Transaction.class); + prepare(mods, tx, false); + + Thread t = new Thread(new Runnable() { + public void run() { + rollback(tx); + } + }); + + t.start(); + t.join(); + + assert !containsKey("k1"); + assert !containsKey("k2"); + assert containsKey("old"); + + mods = new ArrayList<Modification>(); + mods.add(new Store(new StoredEntry("k1", "v1", -1, -1))); + mods.add(new Store(new StoredEntry("k2", "v2", -1, -1))); + mods.add(new Clear()); + mods.add(new Store(new StoredEntry("k3", "v3", -1, -1))); + + prepare(mods, tx, false); + + Thread t2 = new Thread(new Runnable() { + public void run() { + rollback(tx); + } + }); + + t2.start(); + t2.join(); + assert !containsKey("k1"); + assert !containsKey("k2"); + assert !containsKey("k3"); + assert containsKey("old"); + } + + public void testCommitAndRollbackWithoutPrepare() throws CacheLoaderException { + store(new StoredEntry("old", "old", -1, -1)); + Transaction tx = EasyMock.createNiceMock(Transaction.class); + commit(tx); + store(new StoredEntry("old", "old", -1, -1)); + rollback(tx); + + assert containsKey("old"); + } + + public void testPreload() throws CacheLoaderException { + store(new StoredEntry("k1", "v1", -1, -1)); + store(new StoredEntry("k2", "v2", -1, -1)); + store(new StoredEntry("k3", "v3", -1, -1)); + + Set<StoredEntry> set = loadAll(); + + assert set.size() == 3; + Set expected = new HashSet(); + expected.add("k1"); + expected.add("k2"); + expected.add("k3"); + for (StoredEntry se : set) assert expected.remove(se.getKey()); + assert expected.isEmpty(); + } + + @Test + public void testStoreAndRemoveAll() throws CacheLoaderException { + store(new StoredEntry("k1", "v1", -1, -1)); + store(new StoredEntry("k2", "v2", -1, -1)); + store(new StoredEntry("k3", "v3", -1, -1)); + store(new StoredEntry("k4", "v4", -1, -1)); + + + Set<StoredEntry> set = loadAll(); + + assert set.size() == 4; + Set expected = new HashSet(); + expected.add("k1"); + expected.add("k2"); + expected.add("k3"); + expected.add("k4"); + for (StoredEntry se : set) assert expected.remove(se.getKey()); + assert expected.isEmpty(); + + Set toRemove = new HashSet(); + toRemove.add("k1"); + toRemove.add("k2"); + toRemove.add("k3"); + removeAll(toRemove); + + set = loadAll(); + assert set.size() == 1; + set.remove("k4"); + assert expected.isEmpty(); + } + + public void testPurgeExpired() throws Exception { + long now = System.currentTimeMillis(); + long lifespan = 1000; + store(new StoredEntry("k1", "v1", now, now + lifespan)); + store(new StoredEntry("k2", "v2", now, now + lifespan)); + store(new StoredEntry("k3", "v3", now, now + lifespan)); + assert containsKey("k1"); + assert containsKey("k2"); + assert containsKey("k3"); + Thread.sleep(lifespan + 100); + purgeExpired(); + assert !containsKey("k1"); + assert !containsKey("k2"); + assert !containsKey("k3"); + } + + + public void testStreamingAPI() throws IOException, ClassNotFoundException, CacheLoaderException { + store(new StoredEntry("k1", "v1", -1, -1)); + store(new StoredEntry("k2", "v2", -1, -1)); + store(new StoredEntry("k3", "v3", -1, -1)); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(out); + toStream(new UnclosableObjectOutputStream(oos)); + oos.flush(); + oos.close(); + out.close(); + clear(); + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(out.toByteArray())); + fromStream(new UnclosableObjectInputStream(ois)); + + Set<StoredEntry> set = loadAll(); + + assert set.size() == 3; + Set expected = new HashSet(); + expected.add("k1"); + expected.add("k2"); + expected.add("k3"); + for (StoredEntry se : set) assert expected.remove(se.getKey()); + assert expected.isEmpty(); + } + + public void testStreamingAPIReusingStreams() throws IOException, ClassNotFoundException, CacheLoaderException { + store(new StoredEntry("k1", "v1", -1, -1)); + store(new StoredEntry("k2", "v2", -1, -1)); + store(new StoredEntry("k3", "v3", -1, -1)); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] dummyStartBytes = {1, 2, 3, 4, 5, 6, 7, 8}; + byte[] dummyEndBytes = {8, 7, 6, 5, 4, 3, 2, 1}; + out.write(dummyStartBytes); + ObjectOutputStream oos = new ObjectOutputStream(out); + toStream(new UnclosableObjectOutputStream(oos)); + oos.flush(); + oos.close(); + out.write(dummyEndBytes); + out.close(); + clear(); + + // first pop the start bytes + byte[] dummy = new byte[8]; + ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray()); + int bytesRead = in.read(dummy, 0, 8); + assert bytesRead == 8; + for (int i = 1; i < 9; i++) assert dummy[i - 1] == i : "Start byte stream corrupted!"; + fromStream(new UnclosableObjectInputStream(new ObjectInputStream(in))); + bytesRead = in.read(dummy, 0, 8); + assert bytesRead == 8; + for (int i = 8; i > 0; i--) assert dummy[8 - i] == i : "Start byte stream corrupted!"; + + Set<StoredEntry> set = loadAll(); + + assert set.size() == 3; + Set expected = new HashSet(); + expected.add("k1"); + expected.add("k2"); + expected.add("k3"); + for (StoredEntry se : set) assert expected.remove(se.getKey()); + assert expected.isEmpty(); + } + + public void testConcurrency() throws Exception { + int numThreads = 3; + final int loops = 500; + final String[] keys = new String[10]; + final String[] values = new String[10]; + for (int i = 0; i < 10; i++) keys[i] = "k" + i; + for (int i = 0; i < 10; i++) values[i] = "v" + i; + + + final Random r = new Random(); + final List<Exception> exceptions = new LinkedList<Exception>(); + + final Runnable store = new Runnable() { + public void run() { + try { + int randomInt = r.nextInt(10); + store(new StoredEntry(keys[randomInt], values[randomInt])); + } catch (Exception e) { + exceptions.add(e); + } + } + }; + + final Runnable remove = new Runnable() { + public void run() { + try { + remove(keys[r.nextInt(10)]); + } catch (Exception e) { + exceptions.add(e); + } + } + }; + + final Runnable get = new Runnable() { + public void run() { + try { + int randomInt = r.nextInt(10); + StoredEntry se = load(keys[randomInt]); + assert se == null || se.getValue().equals(values[randomInt]); + loadAll(); + } catch (Exception e) { + exceptions.add(e); + } + } + }; + + Thread[] threads = new Thread[numThreads]; + + for (int i = 0; i < numThreads; i++) { + threads[i] = new Thread(getClass().getSimpleName() + "-" + i) { + public void run() { + for (int i = 0; i < loops; i++) { + store.run(); + remove.run(); + get.run(); + } + } + }; + } + + for (Thread t : threads) t.start(); + for (Thread t : threads) t.join(); + + if (!exceptions.isEmpty()) throw exceptions.get(0); + } +}