From 1d71c6979cda451e37e149ae7cc385226d5a3947 Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Thu, 12 Feb 2015 17:57:48 -0500 Subject: [PATCH 1/9] Added a unit test for issue #159 --- .../java/com/simperium/ChannelTest.java | 36 ++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/Simperium/src/androidTestSupport/java/com/simperium/ChannelTest.java b/Simperium/src/androidTestSupport/java/com/simperium/ChannelTest.java index 2d36cc71..8565da7f 100644 --- a/Simperium/src/androidTestSupport/java/com/simperium/ChannelTest.java +++ b/Simperium/src/androidTestSupport/java/com/simperium/ChannelTest.java @@ -99,7 +99,7 @@ protected void tearDown() throws Exception { * Some objects require a certain state before they can be deleted from a bucket so this ensures * the object is at the desired state before the delete operation is sent. */ - public void testSendFinalModificationBeforeDeleteOperation() throws Exception { + public void testQueueFinalModificationBeforeDeleteOperation() throws Exception { mListener.autoAcknowledge = true; @@ -135,6 +135,40 @@ public void testSendFinalModificationBeforeDeleteOperation() throws Exception { } + /** + * See https://github.com/Simperium/simperium-android/issues/159 + * Ensures that the channel is able to send out final modification operations for an object even if + * the object has since been removed from the local persistent store. + */ + public void testSendFinalModificationBeforeDeleteOperation() throws Exception { + + startWithEmptyIndex(); + + String key = "test-modify-before-delete-object"; + Note note = mBucket.newObject(key); + note.setTitle("Bonjour le monde!"); + + note.save(); + + // Queue a deletion and remove the object from the local persistent store before the modification has been sent + note.delete(); + + clearMessages(); + waitForMessage(); + + // Message should be a change message "c:{}" + assertMatchesRegex("^c:\\{.*\\}$", mListener.lastMessage.toString()); + + // First change has been sent and it's a modification + assertEquals(1, mChannelSerializer.queue.pending.size()); + assertTrue(mChannelSerializer.queue.pending.get(key).isModifyOperation()); + + // Second change is queued and it's a deletion + assertEquals(1, mChannelSerializer.queue.queued.size()); + assertTrue(mChannelSerializer.queue.queued.get(0).isRemoveOperation()); + + } + /** * Testing the receipt of cv:? from the server when using api 1.1 */ From b9ab30a6d9f3ecc75ca4d59918d6d4418ce2a2e6 Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Fri, 13 Feb 2015 12:20:54 -0500 Subject: [PATCH 2/9] Modified Bucket to keep a copy of modified objects - Added a method that will return the backup copy if retrieval from persistent store fails --- .../java/com/simperium/client/Bucket.java | 32 +++++++++++++++++-- .../java/com/simperium/client/Channel.java | 2 +- 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/Simperium/src/main/java/com/simperium/client/Bucket.java b/Simperium/src/main/java/com/simperium/client/Bucket.java index 0be5f06d..d01a5a3a 100644 --- a/Simperium/src/main/java/com/simperium/client/Bucket.java +++ b/Simperium/src/main/java/com/simperium/client/Bucket.java @@ -33,8 +33,10 @@ import org.json.JSONObject; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; +import java.util.Map; import java.util.Set; import java.util.concurrent.Executor; @@ -97,6 +99,8 @@ public enum ChangeType { private GhostStorageProvider mGhostStore; final private Executor mExecutor; + private Map mBackupStore = new HashMap<>(); + /** * Represents a Simperium bucket which is a namespace where an app syncs a user's data * @param name the name to use for the bucket namespace @@ -194,6 +198,11 @@ public void run() { Boolean modified = object.isModified(); mStorage.save(object, mSchema.indexesFor(object)); + // TODO: Modify to only clear the HashMap if there are no pending changes for any object in the bucket + mBackupStore.clear(); + // Save a copy in case the object is removed from storage before this modification has been processed + mBackupStore.put(object.getSimperiumKey(), object); + mChannel.queueLocalChange(object); if (modified) { @@ -368,6 +377,25 @@ public T get(String key) throws BucketObjectMissingException { return object; } + /** + * Get an object by its key, checking the backup store if the object has been removed from the persistent store + */ + public T getObjectOrBackup(String key) throws BucketObjectMissingException { + T object; + try { + object = get(key); + } catch (BucketObjectMissingException e) { + // If the object has been removed from the persistent store, check the backup store + object = mBackupStore.get(key); + if (object == null) { + throw(new BucketObjectMissingException(String.format( + "Storage provider for bucket:%s did not have object %s and there was no stored backup", + getName(), key))); + } + } + return object; + } + /** * Get an object by its key, should we throw an error if the object isn't * there? @@ -743,7 +771,7 @@ public Ghost acknowledgeChange(RemoteChange remoteChange, Change change) Ghost ghost = null; if (!remoteChange.isRemoveOperation()) { try { - T object = get(remoteChange.getKey()); + T object = getObjectOrBackup(remoteChange.getKey()); // apply the diff to the underyling object ghost = remoteChange.apply(object.getGhost()); mGhostStore.saveGhost(this, ghost); @@ -780,7 +808,7 @@ public Ghost applyRemoteChange(RemoteChange change) object = newObject(change.getKey()); isNew = true; } else { - object = getObject(change.getKey()); + object = getObjectOrBackup(change.getKey()); isNew = false; notifyOnBeforeUpdateObjectListeners(object); diff --git a/Simperium/src/main/java/com/simperium/client/Channel.java b/Simperium/src/main/java/com/simperium/client/Channel.java index b8dc81b7..41711674 100644 --- a/Simperium/src/main/java/com/simperium/client/Channel.java +++ b/Simperium/src/main/java/com/simperium/client/Channel.java @@ -1484,7 +1484,7 @@ private void sendChange(Change change) try { log(LOG_DEBUG, String.format("Sending change for id: %s op: %s ccid: %s", change.getKey(), change.getOperation(), change.getChangeId())); - Syncable target = mBucket.getObject(change.getKey()); + Syncable target = mBucket.getObjectOrBackup(change.getKey()); Ghost ghost = mBucket.getGhost(change.getKey()); sendMessage(String.format("c:%s", change.toJSONObject(target.getDiffableValue(), ghost))); mSerializer.onSendChange(change); From 55410cc7998391318703871cc9b1f93d3bab140f Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Sun, 15 Feb 2015 11:15:42 -0500 Subject: [PATCH 3/9] Update the backup store when the ghost of one of its entries changes - Fixes an issue found in Simplenote where deleting and trashing a note right after modifying it could cause the deletion not to reach the server. This was due to an object in the backup store having an outdated ghost. --- .../main/java/com/simperium/client/Bucket.java | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/Simperium/src/main/java/com/simperium/client/Bucket.java b/Simperium/src/main/java/com/simperium/client/Bucket.java index d01a5a3a..e6d85c72 100644 --- a/Simperium/src/main/java/com/simperium/client/Bucket.java +++ b/Simperium/src/main/java/com/simperium/client/Bucket.java @@ -99,7 +99,7 @@ public enum ChangeType { private GhostStorageProvider mGhostStore; final private Executor mExecutor; - private Map mBackupStore = new HashMap<>(); + private final Map mBackupStore = new HashMap<>(); /** * Represents a Simperium bucket which is a namespace where an app syncs a user's data @@ -374,6 +374,7 @@ public T get(String key) throws BucketObjectMissingException { Logger.log(TAG, String.format("Fetched ghost for %s %s", key, ghost)); object.setBucket(this); object.setGhost(ghost); + updateBackupStoreGhost(ghost); return object; } @@ -548,6 +549,16 @@ public void run() { }); } + /** + * Update the ghost of an object in the backup store + */ + private void updateBackupStoreGhost(Ghost ghost) { + T object = mBackupStore.get(ghost.getSimperiumKey()); + if (object != null) { + object.setGhost(ghost); + } + } + public Ghost getGhost(String key) throws GhostMissingException { return mGhostStore.getGhost(this, key); } @@ -777,6 +788,7 @@ public Ghost acknowledgeChange(RemoteChange remoteChange, Change change) mGhostStore.saveGhost(this, ghost); // update the object's ghost object.setGhost(ghost); + updateBackupStoreGhost(ghost); } catch (BucketObjectMissingException e) { throw(new RemoteChangeInvalidException(remoteChange, e)); } @@ -831,6 +843,7 @@ public Ghost applyRemoteChange(RemoteChange change) // persist the ghost to storage mGhostStore.saveGhost(this, updatedGhost); object.setGhost(updatedGhost); + updateBackupStoreGhost(updatedGhost); // allow the schema to update the object instance with the new if (isNew) { From 82f81e5b7795d35a8394dd7eacbe92f5b65621fe Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Mon, 16 Feb 2015 10:51:46 -0500 Subject: [PATCH 4/9] Added functionality for automatically clearing the backup store in Bucket - Promoted Channel.isIdle() to a method of the Bucket.Channel interface - Added a unit test for consecutive saving and deleting of two notes (also tests backup store clearing) --- .../java/com/simperium/BucketTest.java | 55 +++++++++++++++++ .../java/com/simperium/client/Bucket.java | 59 +++++++++++++++++-- .../java/com/simperium/test/MockChannel.java | 6 ++ 3 files changed, 116 insertions(+), 4 deletions(-) diff --git a/Simperium/src/androidTestSupport/java/com/simperium/BucketTest.java b/Simperium/src/androidTestSupport/java/com/simperium/BucketTest.java index 8ad68917..3fa81904 100644 --- a/Simperium/src/androidTestSupport/java/com/simperium/BucketTest.java +++ b/Simperium/src/androidTestSupport/java/com/simperium/BucketTest.java @@ -191,4 +191,59 @@ public void testMergeLocalChangesWithUpdatedGhost() assertEquals("Line 1\nLine 2\nLine 3\n", note.getContent()); } + /** + * If two different notes are both saved and then both deleted, they should both be missing from persistent store + * but present in the backup store. + * + * This also checks that the backup store isn't being cleared before the Channel has a chance to retrieve a backup + * object and send it. + */ + public void testConsecutiveSaveDeleteObjects() { + Note note1 = mBucket.newObject(); + note1.setTitle("Hello World"); + + Note note2 = mBucket.newObject(); + note2.setTitle("Hello Again World"); + + note1.save(); + note2.save(); + + note1.delete(); + note2.delete(); + + // Test retrieving notes from persistent store + BucketObjectMissingException note1MissingException = null; + BucketObjectMissingException note2MissingException = null; + try { + mBucket.getObject(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObject(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from persistent store should fail + assertNotNull(note1MissingException); + assertNotNull(note2MissingException); + + // Test retrieving notes from backup store + note1MissingException = null; + note2MissingException = null; + try { + mBucket.getObjectOrBackup(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObjectOrBackup(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from backup store should succeed + assertNull(note1MissingException); + assertNull(note2MissingException); + } + } diff --git a/Simperium/src/main/java/com/simperium/client/Bucket.java b/Simperium/src/main/java/com/simperium/client/Bucket.java index e6d85c72..011ab7a5 100644 --- a/Simperium/src/main/java/com/simperium/client/Bucket.java +++ b/Simperium/src/main/java/com/simperium/client/Bucket.java @@ -49,6 +49,7 @@ public interface Channel { public void start(); public void stop(); public void reset(); + public boolean isIdle(); } public interface OnBeforeUpdateObjectListener { @@ -78,6 +79,9 @@ public enum ChangeType { } public static final String TAG="Simperium.Bucket"; + + private static final int BACKUP_STORE_RESET_DELAY = 5000; + // The name used for the Simperium namespace private String mName; // User provides the access token for authentication @@ -99,7 +103,7 @@ public enum ChangeType { private GhostStorageProvider mGhostStore; final private Executor mExecutor; - private final Map mBackupStore = new HashMap<>(); + private final Map mBackupStore = new TimestampHashMap<>(BACKUP_STORE_RESET_DELAY); /** * Represents a Simperium bucket which is a namespace where an app syncs a user's data @@ -188,6 +192,42 @@ public T getObject() { } + /** + * A HashMap which maintains a timestamp of its last put call and only performs remove(Object) + * and clear() operations if a long enough interval has elapsed. + */ + public class TimestampHashMap extends HashMap { + private long mTimestamp; + private long mClearDelay; + + public TimestampHashMap(long clearDelay) { + mClearDelay = clearDelay; + } + + @Override + public V put(K key, V value) { + V result = super.put(key, value); + mTimestamp = System.currentTimeMillis(); + return result; + } + + @Override + public V remove(Object key) { + V object = null; + if ((System.currentTimeMillis() - mTimestamp) > mClearDelay) { + object = super.remove(key); + } + return object; + } + + @Override + public void clear() { + if ((System.currentTimeMillis() - mTimestamp) > mClearDelay) { + super.clear(); + } + } + } + /** * Tell the bucket to sync changes. */ @@ -198,10 +238,8 @@ public void run() { Boolean modified = object.isModified(); mStorage.save(object, mSchema.indexesFor(object)); - // TODO: Modify to only clear the HashMap if there are no pending changes for any object in the bucket - mBackupStore.clear(); // Save a copy in case the object is removed from storage before this modification has been processed - mBackupStore.put(object.getSimperiumKey(), object); + storeBackupCopy(object); mChannel.queueLocalChange(object); @@ -259,6 +297,19 @@ private void removeObjectWithKey(String key) } } + /** + * Store a copy of the object in the backup store + */ + private void storeBackupCopy(T object) { + synchronized(mBackupStore) { + if (mChannel.isIdle()) { + // If there is no activity in the Channel, we can try to clean up obsolete backups + mBackupStore.clear(); + } + mBackupStore.put(object.getSimperiumKey(), object); + } + } + /** * Get the bucket's namespace * @return (String) bucket's namespace diff --git a/Simperium/src/support/java/com/simperium/test/MockChannel.java b/Simperium/src/support/java/com/simperium/test/MockChannel.java index c23b8e1a..21854582 100644 --- a/Simperium/src/support/java/com/simperium/test/MockChannel.java +++ b/Simperium/src/support/java/com/simperium/test/MockChannel.java @@ -23,6 +23,7 @@ public class MockChannel implements Bucket.Channel { private Bucket mBucket; private boolean started = false; + private boolean idle = false; public boolean autoAcknowledge = true; @@ -41,6 +42,11 @@ public Change queueLocalDeletion(Syncable object){ return change; } + @Override + public boolean isIdle() { + return idle; + } + @Override public Change queueLocalChange(Syncable object) { Change change = new Change(Change.OPERATION_MODIFY, object); From bb85191452a578b5eed6326f7c8e32c8cd4649d2 Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Mon, 16 Feb 2015 16:05:16 -0500 Subject: [PATCH 5/9] Added a thread locking process for Bucket.sync and Bucket.remove - Fixes an issue where calling BucketObject.save() and then immediately BucketObject.remove() on a multi-core device could result in the remove operation being sent to the ChangeProcessor before the modify operation --- .../java/com/simperium/client/Bucket.java | 85 ++++++++++++++++--- 1 file changed, 75 insertions(+), 10 deletions(-) diff --git a/Simperium/src/main/java/com/simperium/client/Bucket.java b/Simperium/src/main/java/com/simperium/client/Bucket.java index 011ab7a5..d9a19c5d 100644 --- a/Simperium/src/main/java/com/simperium/client/Bucket.java +++ b/Simperium/src/main/java/com/simperium/client/Bucket.java @@ -74,6 +74,53 @@ public interface Listener extends // implements all listener methods } + /** + * A one-way exclusion lock that stores multiple keys in a Set. Designed for use with two types of threads, + * primary and secondary, which share a key. + * + * Primary threads control locking by creating a key using lock(E) and deleting it using + * unlock(E). + * + * Secondary threads check whether their key is safe to use using checkLocked(E). If their key is + * present it's not safe to use it, and the threads are blocked until the key is removed from the set. + * + * @param the type of keys maintained + */ + public class LockSet { + private final Set mKeys = new HashSet<>(); + + /** + * Create a new key, blocking secondary threads using that key until unlock(E) is called + */ + public synchronized void lock(E key) { + mKeys.add(key); + } + + /** + * Block calling thread until lock is released + */ + public synchronized void checkLocked(E key) throws InterruptedException { + while (mKeys.contains(key)) { + wait(); + } + } + + /** + * Remove a key from the map, releasing the lock + */ + public synchronized void unlock(E key){ + mKeys.remove(key); + notify(); + } + + /** + * Return lock status for given key + */ + public synchronized boolean isLocked(E key){ + return mKeys.contains(key); + } + } + public enum ChangeType { REMOVE, MODIFY, INDEX, RESET, INSERT } @@ -102,9 +149,10 @@ public enum ChangeType { private BucketSchema mSchema; private GhostStorageProvider mGhostStore; final private Executor mExecutor; - private final Map mBackupStore = new TimestampHashMap<>(BACKUP_STORE_RESET_DELAY); + private final LockSet mSaveDeleteLock = new LockSet<>(); + /** * Represents a Simperium bucket which is a namespace where an app syncs a user's data * @param name the name to use for the bucket namespace @@ -232,21 +280,26 @@ public void clear() { * Tell the bucket to sync changes. */ public void sync(final T object) { + mSaveDeleteLock.lock(object.getSimperiumKey()); mExecutor.execute(new Runnable() { @Override public void run() { - Boolean modified = object.isModified(); - mStorage.save(object, mSchema.indexesFor(object)); + try { + Boolean modified = object.isModified(); + mStorage.save(object, mSchema.indexesFor(object)); - // Save a copy in case the object is removed from storage before this modification has been processed - storeBackupCopy(object); + // Save a copy in case the object is removed from storage before this modification has been processed + storeBackupCopy(object); - mChannel.queueLocalChange(object); + mChannel.queueLocalChange(object); - if (modified) { - // Notify listeners that an object has been saved, this was - // triggered locally - notifyOnSaveListeners(object); + if (modified) { + // Notify listeners that an object has been saved, this was + // triggered locally + notifyOnSaveListeners(object); + } + } finally { + mSaveDeleteLock.unlock(object.getSimperiumKey()); } } }); @@ -273,6 +326,17 @@ private void remove(final T object, final boolean isLocal) { @Override public void run() { if (isLocal) { + try { + mSaveDeleteLock.checkLocked(object.getSimperiumKey()); + } catch (InterruptedException e) { + Logger.log(TAG, String.format("Delete lock interrupted for %s, did not remove from storage", + object.getSimperiumKey())); + // Proceed, but don't remove from storage + mChannel.queueLocalDeletion(object); + notifyOnDeleteListeners(object); + return; + } + mChannel.queueLocalDeletion(object); } @@ -444,6 +508,7 @@ public T getObjectOrBackup(String key) throws BucketObjectMissingException { "Storage provider for bucket:%s did not have object %s and there was no stored backup", getName(), key))); } + Logger.log(TAG, String.format("Fetched backup copy for %s", key)); } return object; } From 2d6518297afb38903947645c755ef17308c7809f Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Mon, 16 Feb 2015 16:48:47 -0500 Subject: [PATCH 6/9] Added a unit test class for testing on a multi-threaded Executor --- .../java/com/simperium/ConcurrencyTest.java | 334 ++++++++++++++++++ 1 file changed, 334 insertions(+) create mode 100644 Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java diff --git a/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java new file mode 100644 index 00000000..379fab43 --- /dev/null +++ b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java @@ -0,0 +1,334 @@ +package com.simperium; + +import com.simperium.client.Bucket; +import com.simperium.client.BucketObjectMissingException; +import com.simperium.client.BucketSchema; +import com.simperium.client.Channel; +import com.simperium.client.GhostStorageProvider; +import com.simperium.client.User; +import com.simperium.models.Note; +import com.simperium.storage.MemoryStore; +import com.simperium.test.MockChannelListener; +import com.simperium.test.MockChannelSerializer; +import com.simperium.test.MockGhostStore; +import com.simperium.util.Logger; + +import org.json.JSONArray; +import org.json.JSONObject; + +import java.util.Map; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; + +import static android.test.MoreAsserts.assertMatchesRegex; +import static com.simperium.TestHelpers.makeUser; +import static com.simperium.TestHelpers.waitUntil; + +public class ConcurrencyTest extends BaseSimperiumTest { + public static final String TAG = "SimperiumTest"; + + private static String BUCKET_NAME = "concurrency-test"; + public static String SESSION_ID = "SESSION-ID"; + public static String APP_ID = "APP_ID"; + + private Bucket mBucket; + private Channel mChannel; + private MockChannelSerializer mChannelSerializer = new MockChannelSerializer(); + final private MockChannelListener mListener = new MockChannelListener(); + + private BucketSchema mSchema; + private User mUser; + private MemoryStore mStorage; + private GhostStorageProvider mGhostStore; + + protected User.Status mAuthStatus; + + private Executor mExecutor; + + /** + * Build Bucket and Channel instances using a multi-threaded Executor + */ + protected void setUp() throws Exception { + super.setUp(); + + // Mimic AndroidClient's Executor setup + int threads = Runtime.getRuntime().availableProcessors(); + if (threads > 1) { + threads -= 1; + } else { + Logger.log(TAG, "Single-threaded. ConcurrencyTest won't give any new information."); + } + + mExecutor = Executors.newFixedThreadPool(threads); + + mUser = makeUser(); + mSchema = new Note.Schema(); + mStorage = new MemoryStore(); + mGhostStore = new MockGhostStore(); + mBucket = new Bucket<>(mExecutor, BUCKET_NAME, mSchema, mUser, mStorage.createStore(BUCKET_NAME, mSchema), mGhostStore); + + mChannel = new Channel(mExecutor, APP_ID, SESSION_ID, mBucket, mChannelSerializer, mListener); + mBucket.setChannel(mChannel); + + mBucket.getUser().setStatusChangeListener(new User.StatusChangeListener(){ + @Override + public void onUserStatusChange(User.Status status){ + mAuthStatus = status; + } + }); + + } + + protected void tearDown() throws Exception { + mChannel.stop(); + mChannel.reset(); + super.tearDown(); + } + + /** + * Same as ChannelTest.testSendFinalModificationBeforeDeleteOperation() but with concurrency enabled. + * + * See https://github.com/Simperium/simperium-android/issues/159 + * Ensures that the channel is able to send out final modification operations for an object even if + * the object has since been removed from the local persistent store. + */ + public void testSendFinalModificationBeforeDeleteOperationConcurrent() throws Exception { + + startWithEmptyIndex(); + + String key = "test-modify-before-delete-object"; + Note note = mBucket.newObject(key); + note.setTitle("Bonjour le monde!"); + + note.save(); + + // Queue a deletion and remove the object from the local persistent store before the modification has been sent + note.delete(); + + clearMessages(); + waitForMessage(); + + // Message should be a change message "c:{}" + assertMatchesRegex("^c:\\{.*\\}$", mListener.lastMessage.toString()); + + // First change has been sent and it's a modification + assertEquals(1, mChannelSerializer.queue.pending.size()); + assertTrue(mChannelSerializer.queue.pending.get(key).isModifyOperation()); + + // Second change is queued and it's a deletion + assertEquals(1, mChannelSerializer.queue.queued.size()); + assertTrue(mChannelSerializer.queue.queued.get(0).isRemoveOperation()); + + } + +// public void testSendFinalModificationsBeforeDeleteOperation() throws Exception { +// +// startWithEmptyIndex(); +// +// String key = "test-multi-save"; +// Note note = mBucket.newObject(key); +// +// note.setTitle("title1"); +// note.save(); +// +// note.setContent("another name"); +// note.save(); +// +// // Queue a deletion and remove the object from the local persistent store before the modification has been sent +// note.delete(); +// +// clearMessages(); +// waitForMessage(); +// +// // Message should be a change message "c:{}" +// Logger.log("WHEE", mListener.lastMessage.toString()); +// assertMatchesRegex("^c:\\{.*\\}$", mListener.lastMessage.toString()); +// +//// // First change has been sent and it's a modification +//// assertEquals(1, mChannelSerializer.queue.pending.size()); +//// assertTrue(mChannelSerializer.queue.pending.get(key).isModifyOperation()); +//// +//// // Second change is queued and it's a deletion +//// assertEquals(1, mChannelSerializer.queue.queued.size()); +//// assertTrue(mChannelSerializer.queue.queued.get(0).isRemoveOperation()); +// +// } + + /** + * Same as BucketTest.testConsecutiveSaveDeleteObjects() but with concurrency enabled. + * + * If two different notes are both saved and then both deleted, they should both be missing from persistent store + * but present in the backup store. + * + * This also checks that the backup store isn't being cleared before the Channel has a chance to retrieve a backup + * object and send it. + */ + public void testConsecutiveSaveDeleteObjectsConcurrent() { + Note note1 = mBucket.newObject(); + note1.setTitle("Hello World"); + + Note note2 = mBucket.newObject(); + note2.setTitle("Hello Again World"); + + note1.save(); + note2.save(); + + note1.delete(); + note2.delete(); + + // Allow the save threads to finish writing to storage before querying it + waitFor(10); + + // Test retrieving notes from persistent store + BucketObjectMissingException note1MissingException = null; + BucketObjectMissingException note2MissingException = null; + try { + mBucket.getObject(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObject(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from persistent store should fail + assertNotNull(note1MissingException); + assertNotNull(note2MissingException); + + // Test retrieving notes from backup store + note1MissingException = null; + note2MissingException = null; + try { + mBucket.getObjectOrBackup(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObjectOrBackup(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from backup store should succeed + assertNull(note1MissingException); + assertNull(note2MissingException); + } + + + /** + * Gets the channel into a started state + */ + protected void start(){ + mChannel.onConnect(); + mChannel.start(); + // send auth success message + mChannel.receiveMessage("auth:user@example.com"); + } + + protected void startWithEmptyIndex() + throws Exception { + start(); + sendEmptyIndex(); + } + + protected void startWithIndex(Map objects) + throws Exception { + startWithIndex("mock-cv", objects); + } + + protected void startWithIndex(String cv, Map objects) + throws Exception { + start(); + sendIndex(cv, objects); + } + + /** + * Simulates a brand new bucket with and empty index + */ + protected void sendEmptyIndex() + throws Exception { + sendMessage("i:{\"index\":[]}"); + waitForIndex(); + } + + protected void sendIndex(String cv, Map objects) + throws Exception { + JSONObject index = new JSONObject(); + String period = "."; + index.put("current", cv); + + JSONArray versions = new JSONArray(); + index.put("index", versions); + + for(Map.Entry entry : objects.entrySet()) { + String key = entry.getKey(); + int dot = key.indexOf(period); + String id = key.substring(0, dot); + int version = Integer.parseInt(key.substring(dot+1)); + JSONObject versionData = new JSONObject(); + versionData.put("v", version); + versionData.put("id", id); + + versions.put(versionData); + } + + mListener.indexVersions = versions; + mListener.indexData = objects; + sendMessage(String.format("i:%s", index)); + waitForIndex(); + } + + protected void sendMessage(String message){ + mChannel.receiveMessage(message); + } + + protected Channel.MessageEvent waitForMessage() throws InterruptedException { + return waitForMessage(2000); + } + + /** + * Wait until a message received. More than likely clearMessages() should + * be called before waitForMessage() + */ + protected Channel.MessageEvent waitForMessage(int waitFor) throws InterruptedException { + + NewMessageFlagger flagger = new NewMessageFlagger(); + + waitUntil(flagger, "No message received", waitFor); + + return flagger.message; + + } + + /** + * Empties the list of received messages and sets last message to null + */ + protected void clearMessages(){ + mListener.clearMessages(); + } + + protected void waitForIndex() + throws InterruptedException { + waitUntil(new TestHelpers.Flag(){ + @Override + public boolean isComplete(){ + return mChannel.haveCompleteIndex(); + } + }, "Index never received", 5000); + } + + private class NewMessageFlagger implements TestHelpers.Flag { + + Channel.MessageEvent message; + + @Override + public boolean isComplete() { + + message = mListener.lastMessage; + + return mListener.lastMessage != null; + + } + + } +} From 3510f6715f64ffd8c04831731731b54a3c95e15d Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Mon, 16 Feb 2015 17:00:07 -0500 Subject: [PATCH 7/9] Added a unit test class for testing on a multi-threaded Executor --- .../java/com/simperium/ConcurrencyTest.java | 334 ++++++++++++++++++ 1 file changed, 334 insertions(+) create mode 100644 Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java diff --git a/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java new file mode 100644 index 00000000..379fab43 --- /dev/null +++ b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java @@ -0,0 +1,334 @@ +package com.simperium; + +import com.simperium.client.Bucket; +import com.simperium.client.BucketObjectMissingException; +import com.simperium.client.BucketSchema; +import com.simperium.client.Channel; +import com.simperium.client.GhostStorageProvider; +import com.simperium.client.User; +import com.simperium.models.Note; +import com.simperium.storage.MemoryStore; +import com.simperium.test.MockChannelListener; +import com.simperium.test.MockChannelSerializer; +import com.simperium.test.MockGhostStore; +import com.simperium.util.Logger; + +import org.json.JSONArray; +import org.json.JSONObject; + +import java.util.Map; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; + +import static android.test.MoreAsserts.assertMatchesRegex; +import static com.simperium.TestHelpers.makeUser; +import static com.simperium.TestHelpers.waitUntil; + +public class ConcurrencyTest extends BaseSimperiumTest { + public static final String TAG = "SimperiumTest"; + + private static String BUCKET_NAME = "concurrency-test"; + public static String SESSION_ID = "SESSION-ID"; + public static String APP_ID = "APP_ID"; + + private Bucket mBucket; + private Channel mChannel; + private MockChannelSerializer mChannelSerializer = new MockChannelSerializer(); + final private MockChannelListener mListener = new MockChannelListener(); + + private BucketSchema mSchema; + private User mUser; + private MemoryStore mStorage; + private GhostStorageProvider mGhostStore; + + protected User.Status mAuthStatus; + + private Executor mExecutor; + + /** + * Build Bucket and Channel instances using a multi-threaded Executor + */ + protected void setUp() throws Exception { + super.setUp(); + + // Mimic AndroidClient's Executor setup + int threads = Runtime.getRuntime().availableProcessors(); + if (threads > 1) { + threads -= 1; + } else { + Logger.log(TAG, "Single-threaded. ConcurrencyTest won't give any new information."); + } + + mExecutor = Executors.newFixedThreadPool(threads); + + mUser = makeUser(); + mSchema = new Note.Schema(); + mStorage = new MemoryStore(); + mGhostStore = new MockGhostStore(); + mBucket = new Bucket<>(mExecutor, BUCKET_NAME, mSchema, mUser, mStorage.createStore(BUCKET_NAME, mSchema), mGhostStore); + + mChannel = new Channel(mExecutor, APP_ID, SESSION_ID, mBucket, mChannelSerializer, mListener); + mBucket.setChannel(mChannel); + + mBucket.getUser().setStatusChangeListener(new User.StatusChangeListener(){ + @Override + public void onUserStatusChange(User.Status status){ + mAuthStatus = status; + } + }); + + } + + protected void tearDown() throws Exception { + mChannel.stop(); + mChannel.reset(); + super.tearDown(); + } + + /** + * Same as ChannelTest.testSendFinalModificationBeforeDeleteOperation() but with concurrency enabled. + * + * See https://github.com/Simperium/simperium-android/issues/159 + * Ensures that the channel is able to send out final modification operations for an object even if + * the object has since been removed from the local persistent store. + */ + public void testSendFinalModificationBeforeDeleteOperationConcurrent() throws Exception { + + startWithEmptyIndex(); + + String key = "test-modify-before-delete-object"; + Note note = mBucket.newObject(key); + note.setTitle("Bonjour le monde!"); + + note.save(); + + // Queue a deletion and remove the object from the local persistent store before the modification has been sent + note.delete(); + + clearMessages(); + waitForMessage(); + + // Message should be a change message "c:{}" + assertMatchesRegex("^c:\\{.*\\}$", mListener.lastMessage.toString()); + + // First change has been sent and it's a modification + assertEquals(1, mChannelSerializer.queue.pending.size()); + assertTrue(mChannelSerializer.queue.pending.get(key).isModifyOperation()); + + // Second change is queued and it's a deletion + assertEquals(1, mChannelSerializer.queue.queued.size()); + assertTrue(mChannelSerializer.queue.queued.get(0).isRemoveOperation()); + + } + +// public void testSendFinalModificationsBeforeDeleteOperation() throws Exception { +// +// startWithEmptyIndex(); +// +// String key = "test-multi-save"; +// Note note = mBucket.newObject(key); +// +// note.setTitle("title1"); +// note.save(); +// +// note.setContent("another name"); +// note.save(); +// +// // Queue a deletion and remove the object from the local persistent store before the modification has been sent +// note.delete(); +// +// clearMessages(); +// waitForMessage(); +// +// // Message should be a change message "c:{}" +// Logger.log("WHEE", mListener.lastMessage.toString()); +// assertMatchesRegex("^c:\\{.*\\}$", mListener.lastMessage.toString()); +// +//// // First change has been sent and it's a modification +//// assertEquals(1, mChannelSerializer.queue.pending.size()); +//// assertTrue(mChannelSerializer.queue.pending.get(key).isModifyOperation()); +//// +//// // Second change is queued and it's a deletion +//// assertEquals(1, mChannelSerializer.queue.queued.size()); +//// assertTrue(mChannelSerializer.queue.queued.get(0).isRemoveOperation()); +// +// } + + /** + * Same as BucketTest.testConsecutiveSaveDeleteObjects() but with concurrency enabled. + * + * If two different notes are both saved and then both deleted, they should both be missing from persistent store + * but present in the backup store. + * + * This also checks that the backup store isn't being cleared before the Channel has a chance to retrieve a backup + * object and send it. + */ + public void testConsecutiveSaveDeleteObjectsConcurrent() { + Note note1 = mBucket.newObject(); + note1.setTitle("Hello World"); + + Note note2 = mBucket.newObject(); + note2.setTitle("Hello Again World"); + + note1.save(); + note2.save(); + + note1.delete(); + note2.delete(); + + // Allow the save threads to finish writing to storage before querying it + waitFor(10); + + // Test retrieving notes from persistent store + BucketObjectMissingException note1MissingException = null; + BucketObjectMissingException note2MissingException = null; + try { + mBucket.getObject(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObject(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from persistent store should fail + assertNotNull(note1MissingException); + assertNotNull(note2MissingException); + + // Test retrieving notes from backup store + note1MissingException = null; + note2MissingException = null; + try { + mBucket.getObjectOrBackup(note1.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note1MissingException = e; + } + try { + mBucket.getObjectOrBackup(note2.getSimperiumKey()); + } catch (BucketObjectMissingException e) { + note2MissingException = e; + } + // Retrieval from backup store should succeed + assertNull(note1MissingException); + assertNull(note2MissingException); + } + + + /** + * Gets the channel into a started state + */ + protected void start(){ + mChannel.onConnect(); + mChannel.start(); + // send auth success message + mChannel.receiveMessage("auth:user@example.com"); + } + + protected void startWithEmptyIndex() + throws Exception { + start(); + sendEmptyIndex(); + } + + protected void startWithIndex(Map objects) + throws Exception { + startWithIndex("mock-cv", objects); + } + + protected void startWithIndex(String cv, Map objects) + throws Exception { + start(); + sendIndex(cv, objects); + } + + /** + * Simulates a brand new bucket with and empty index + */ + protected void sendEmptyIndex() + throws Exception { + sendMessage("i:{\"index\":[]}"); + waitForIndex(); + } + + protected void sendIndex(String cv, Map objects) + throws Exception { + JSONObject index = new JSONObject(); + String period = "."; + index.put("current", cv); + + JSONArray versions = new JSONArray(); + index.put("index", versions); + + for(Map.Entry entry : objects.entrySet()) { + String key = entry.getKey(); + int dot = key.indexOf(period); + String id = key.substring(0, dot); + int version = Integer.parseInt(key.substring(dot+1)); + JSONObject versionData = new JSONObject(); + versionData.put("v", version); + versionData.put("id", id); + + versions.put(versionData); + } + + mListener.indexVersions = versions; + mListener.indexData = objects; + sendMessage(String.format("i:%s", index)); + waitForIndex(); + } + + protected void sendMessage(String message){ + mChannel.receiveMessage(message); + } + + protected Channel.MessageEvent waitForMessage() throws InterruptedException { + return waitForMessage(2000); + } + + /** + * Wait until a message received. More than likely clearMessages() should + * be called before waitForMessage() + */ + protected Channel.MessageEvent waitForMessage(int waitFor) throws InterruptedException { + + NewMessageFlagger flagger = new NewMessageFlagger(); + + waitUntil(flagger, "No message received", waitFor); + + return flagger.message; + + } + + /** + * Empties the list of received messages and sets last message to null + */ + protected void clearMessages(){ + mListener.clearMessages(); + } + + protected void waitForIndex() + throws InterruptedException { + waitUntil(new TestHelpers.Flag(){ + @Override + public boolean isComplete(){ + return mChannel.haveCompleteIndex(); + } + }, "Index never received", 5000); + } + + private class NewMessageFlagger implements TestHelpers.Flag { + + Channel.MessageEvent message; + + @Override + public boolean isComplete() { + + message = mListener.lastMessage; + + return mListener.lastMessage != null; + + } + + } +} From 76379fd7d3d018164d1e125cf9bc001d648960da Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Mon, 16 Feb 2015 17:02:20 -0500 Subject: [PATCH 8/9] Cleaned up some obsolete code --- .../java/com/simperium/ConcurrencyTest.java | 33 ------------------- 1 file changed, 33 deletions(-) diff --git a/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java index 379fab43..0678abfb 100644 --- a/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java +++ b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java @@ -121,39 +121,6 @@ public void testSendFinalModificationBeforeDeleteOperationConcurrent() throws Ex } -// public void testSendFinalModificationsBeforeDeleteOperation() throws Exception { -// -// startWithEmptyIndex(); -// -// String key = "test-multi-save"; -// Note note = mBucket.newObject(key); -// -// note.setTitle("title1"); -// note.save(); -// -// note.setContent("another name"); -// note.save(); -// -// // Queue a deletion and remove the object from the local persistent store before the modification has been sent -// note.delete(); -// -// clearMessages(); -// waitForMessage(); -// -// // Message should be a change message "c:{}" -// Logger.log("WHEE", mListener.lastMessage.toString()); -// assertMatchesRegex("^c:\\{.*\\}$", mListener.lastMessage.toString()); -// -//// // First change has been sent and it's a modification -//// assertEquals(1, mChannelSerializer.queue.pending.size()); -//// assertTrue(mChannelSerializer.queue.pending.get(key).isModifyOperation()); -//// -//// // Second change is queued and it's a deletion -//// assertEquals(1, mChannelSerializer.queue.queued.size()); -//// assertTrue(mChannelSerializer.queue.queued.get(0).isRemoveOperation()); -// -// } - /** * Same as BucketTest.testConsecutiveSaveDeleteObjects() but with concurrency enabled. * From 746b558f4ff862bc79164f77806cbc2cb8c11f99 Mon Sep 17 00:00:00 2001 From: Alex Forcier Date: Mon, 16 Feb 2015 20:30:50 -0500 Subject: [PATCH 9/9] Updated a concurrency test to wait for Executor tasks to complete rather than wait a fixed time --- .../java/com/simperium/ConcurrencyTest.java | 23 ++++++++++++------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java index 0678abfb..2342370e 100644 --- a/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java +++ b/Simperium/src/androidTestSupport/java/com/simperium/ConcurrencyTest.java @@ -17,8 +17,8 @@ import org.json.JSONObject; import java.util.Map; -import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import java.util.concurrent.ThreadPoolExecutor; import static android.test.MoreAsserts.assertMatchesRegex; import static com.simperium.TestHelpers.makeUser; @@ -43,7 +43,7 @@ public class ConcurrencyTest extends BaseSimperiumTest { protected User.Status mAuthStatus; - private Executor mExecutor; + private ThreadPoolExecutor mExecutor; /** * Build Bucket and Channel instances using a multi-threaded Executor @@ -55,11 +55,9 @@ protected void setUp() throws Exception { int threads = Runtime.getRuntime().availableProcessors(); if (threads > 1) { threads -= 1; - } else { - Logger.log(TAG, "Single-threaded. ConcurrencyTest won't give any new information."); } - mExecutor = Executors.newFixedThreadPool(threads); + mExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(threads); mUser = makeUser(); mSchema = new Note.Schema(); @@ -130,7 +128,7 @@ public void testSendFinalModificationBeforeDeleteOperationConcurrent() throws Ex * This also checks that the backup store isn't being cleared before the Channel has a chance to retrieve a backup * object and send it. */ - public void testConsecutiveSaveDeleteObjectsConcurrent() { + public void testConsecutiveSaveDeleteObjectsConcurrent() throws InterruptedException { Note note1 = mBucket.newObject(); note1.setTitle("Hello World"); @@ -143,8 +141,8 @@ public void testConsecutiveSaveDeleteObjectsConcurrent() { note1.delete(); note2.delete(); - // Allow the save threads to finish writing to storage before querying it - waitFor(10); + // Allow the save and delete tasks to finish before querying storage + waitForExecutorCompletedTasks(4); // Test retrieving notes from persistent store BucketObjectMissingException note1MissingException = null; @@ -284,6 +282,15 @@ public boolean isComplete(){ }, "Index never received", 5000); } + protected void waitForExecutorCompletedTasks(final int completedTasks) throws InterruptedException { + waitUntil(new TestHelpers.Flag(){ + @Override + public boolean isComplete(){ + return (mExecutor.getCompletedTaskCount() == completedTasks); + } + }, "Completed task amount never reached", 5000); + } + private class NewMessageFlagger implements TestHelpers.Flag { Channel.MessageEvent message;