You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by cs...@apache.org on 2016/04/12 00:15:33 UTC
[1/2] incubator-apex-malhar git commit: APEXMALHAR-2042 #resolve
#comment fixed race condition where state could be cleared before data is
transfered to a file.
Repository: incubator-apex-malhar
Updated Branches:
refs/heads/master 4cffeb344 -> c4052f8c6
APEXMALHAR-2042 #resolve #comment fixed race condition where state could be cleared before data is transfered to a file.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/1d3d8763
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/1d3d8763
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/1d3d8763
Branch: refs/heads/master
Commit: 1d3d8763d4a7552396991b391f8a23ab214fb65b
Parents: 4cffeb3
Author: Timothy Farkas <ti...@datatorrent.com>
Authored: Thu Apr 7 14:27:39 2016 -0700
Committer: Timothy Farkas <ti...@datatorrent.com>
Committed: Mon Apr 11 15:02:17 2016 -0700
----------------------------------------------------------------------
.../apex/malhar/lib/state/managed/Bucket.java | 42 ++++++++++++-----
.../managed/IncrementalCheckpointManager.java | 16 +++++--
.../malhar/lib/state/managed/StateTracker.java | 2 +-
.../lib/state/managed/DefaultBucketTest.java | 2 +-
.../lib/state/managed/ManagedStateImplTest.java | 49 ++++++++++++++++++--
.../lib/state/managed/StateTrackerTest.java | 4 +-
6 files changed, 94 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/1d3d8763/library/src/main/java/org/apache/apex/malhar/lib/state/managed/Bucket.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/state/managed/Bucket.java b/library/src/main/java/org/apache/apex/malhar/lib/state/managed/Bucket.java
index b2c1618..107bfc6 100644
--- a/library/src/main/java/org/apache/apex/malhar/lib/state/managed/Bucket.java
+++ b/library/src/main/java/org/apache/apex/malhar/lib/state/managed/Bucket.java
@@ -92,10 +92,12 @@ public interface Bucket extends ManagedStateComponent
/**
* Triggers bucket to free memory which is already persisted in bucket data files.
*
+ * @param windowId All the data corresponding to windowId's less than or equal to this winowId
+ * will be freed.
* @return amount of memory freed in bytes.
* @throws IOException
*/
- long freeMemory() throws IOException;
+ long freeMemory(long windowId) throws IOException;
/**
* Allows the bucket to process/cache data which is recovered (from window files) after failure.
@@ -186,7 +188,8 @@ public interface Bucket extends ManagedStateComponent
new ConcurrentSkipListMap<>();
//Data persisted in bucket data files
- private final transient Map<Slice, BucketedValue> committedData = Maps.newConcurrentMap();
+ private final transient ConcurrentSkipListMap<Long, Map<Slice, BucketedValue>> committedData =
+ new ConcurrentSkipListMap<>();
//Data serialized/deserialized from bucket data files: key -> value from latest time bucket on file
private final transient Map<Slice, BucketedValue> fileCache = Maps.newConcurrentMap();
@@ -247,9 +250,12 @@ public interface Bucket extends ManagedStateComponent
}
}
- bucketedValue = committedData.get(key);
- if (bucketedValue != null) {
- return bucketedValue.getValue();
+ for (Long window : committedData.descendingKeySet()) {
+ //traverse the committed data in reverse order
+ bucketedValue = committedData.get(window).get(key);
+ if (bucketedValue != null) {
+ return bucketedValue.getValue();
+ }
}
bucketedValue = fileCache.get(key);
@@ -392,14 +398,27 @@ public interface Bucket extends ManagedStateComponent
}
@Override
- public long freeMemory() throws IOException
+ public long freeMemory(long windowId) throws IOException
{
LOG.debug("free space {}", bucketId);
long memoryFreed = 0;
- for (Map.Entry<Slice, BucketedValue> entry : committedData.entrySet()) {
- memoryFreed += entry.getKey().length + entry.getValue().getValue().length;
+
+ for (Map.Entry<Long, Map<Slice, BucketedValue>> windowEntry : committedData.entrySet()) {
+ for (Map.Entry<Slice, BucketedValue> entry: windowEntry.getValue().entrySet()) {
+ memoryFreed += entry.getKey().length + entry.getValue().getValue().length;
+ }
}
- committedData.clear();
+
+ Long clearWindowId = null;
+
+ while ((clearWindowId = committedData.floorKey(windowId)) != null) {
+ Map<Slice, BucketedValue> windowData = committedData.remove(clearWindowId);
+
+ for (Map.Entry<Slice, BucketedValue> entry: windowData.entrySet()) {
+ memoryFreed += entry.getKey().length + entry.getValue().getValue().length;
+ }
+ }
+
fileCache.clear();
if (cachedBucketMetas != null) {
@@ -462,7 +481,8 @@ public interface Bucket extends ManagedStateComponent
break;
}
}
- committedData.putAll(bucketData);
+
+ committedData.put(savedWindow, bucketData);
stateIterator.remove();
} else {
break;
@@ -509,7 +529,7 @@ public interface Bucket extends ManagedStateComponent
}
@VisibleForTesting
- Map<Slice, BucketedValue> getCommittedData()
+ ConcurrentSkipListMap<Long, Map<Slice, BucketedValue>> getCommittedData()
{
return committedData;
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/1d3d8763/library/src/main/java/org/apache/apex/malhar/lib/state/managed/IncrementalCheckpointManager.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/state/managed/IncrementalCheckpointManager.java b/library/src/main/java/org/apache/apex/malhar/lib/state/managed/IncrementalCheckpointManager.java
index a372163..c7ae1c1 100644
--- a/library/src/main/java/org/apache/apex/malhar/lib/state/managed/IncrementalCheckpointManager.java
+++ b/library/src/main/java/org/apache/apex/malhar/lib/state/managed/IncrementalCheckpointManager.java
@@ -37,6 +37,7 @@ import com.google.common.base.Throwables;
import com.google.common.collect.Queues;
import com.datatorrent.api.Context;
+import com.datatorrent.api.annotation.Stateless;
import com.datatorrent.common.util.NameableThreadFactory;
import com.datatorrent.lib.util.WindowDataManager;
import com.datatorrent.netlet.util.Slice;
@@ -67,7 +68,7 @@ public class IncrementalCheckpointManager extends WindowDataManager.FSWindowData
private final transient AtomicLong latestExpiredTimeBucket = new AtomicLong(-1);
private transient int waitMillis;
-
+ private volatile long lastTransferredWindow = Stateless.WINDOW_ID;
public IncrementalCheckpointManager()
{
@@ -132,6 +133,8 @@ public class IncrementalCheckpointManager extends WindowDataManager.FSWindowData
LOG.debug("transfer window {}", windowId, t);
Throwables.propagate(t);
}
+
+ this.lastTransferredWindow = windowId;
} else {
Thread.sleep(waitMillis);
}
@@ -173,8 +176,6 @@ public class IncrementalCheckpointManager extends WindowDataManager.FSWindowData
}
}
-
-
/**
* Transfers the data which has been committed till windowId to data files.
*
@@ -208,6 +209,15 @@ public class IncrementalCheckpointManager extends WindowDataManager.FSWindowData
latestExpiredTimeBucket.set(timeBucket);
}
+ /**
+ * Gets the last windowId for which data was successfully merged with a bucket data file.
+ * @return The last windowId for which data was successfully merged with a bucket data file.
+ */
+ public long getLastTransferredWindow()
+ {
+ return lastTransferredWindow;
+ }
+
private static final Logger LOG = LoggerFactory.getLogger(IncrementalCheckpointManager.class);
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/1d3d8763/library/src/main/java/org/apache/apex/malhar/lib/state/managed/StateTracker.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/state/managed/StateTracker.java b/library/src/main/java/org/apache/apex/malhar/lib/state/managed/StateTracker.java
index a359b31..3bb5507 100644
--- a/library/src/main/java/org/apache/apex/malhar/lib/state/managed/StateTracker.java
+++ b/library/src/main/java/org/apache/apex/malhar/lib/state/managed/StateTracker.java
@@ -122,7 +122,7 @@ class StateTracker extends TimerTask
synchronized (bucket) {
long sizeFreed;
try {
- sizeFreed = bucket.freeMemory();
+ sizeFreed = bucket.freeMemory(managedStateImpl.checkpointManager.getLastTransferredWindow());
LOG.debug("size freed {} {}", bucketId, sizeFreed);
} catch (IOException e) {
managedStateImpl.throwable.set(e);
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/1d3d8763/library/src/test/java/org/apache/apex/malhar/lib/state/managed/DefaultBucketTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/org/apache/apex/malhar/lib/state/managed/DefaultBucketTest.java b/library/src/test/java/org/apache/apex/malhar/lib/state/managed/DefaultBucketTest.java
index 93f705b..7539440 100644
--- a/library/src/test/java/org/apache/apex/malhar/lib/state/managed/DefaultBucketTest.java
+++ b/library/src/test/java/org/apache/apex/malhar/lib/state/managed/DefaultBucketTest.java
@@ -195,7 +195,7 @@ public class DefaultBucketTest
Assert.assertEquals("size", initSize + (two.length * 2 + 64), testMeta.defaultBucket.getSizeInBytes());
- long sizeFreed = testMeta.defaultBucket.freeMemory();
+ long sizeFreed = testMeta.defaultBucket.freeMemory(Long.MAX_VALUE);
Assert.assertEquals("size freed", initSize, sizeFreed);
Assert.assertEquals("existing size", (two.length * 2 + 64), testMeta.defaultBucket.getSizeInBytes());
testMeta.defaultBucket.teardown();
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/1d3d8763/library/src/test/java/org/apache/apex/malhar/lib/state/managed/ManagedStateImplTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/org/apache/apex/malhar/lib/state/managed/ManagedStateImplTest.java b/library/src/test/java/org/apache/apex/malhar/lib/state/managed/ManagedStateImplTest.java
index bc8b1e8..cdd9781 100644
--- a/library/src/test/java/org/apache/apex/malhar/lib/state/managed/ManagedStateImplTest.java
+++ b/library/src/test/java/org/apache/apex/malhar/lib/state/managed/ManagedStateImplTest.java
@@ -23,11 +23,14 @@ import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
+import org.joda.time.Duration;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TestWatcher;
import org.junit.runner.Description;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.datatorrent.api.Context;
import com.datatorrent.lib.fileaccess.FileAccessFSImpl;
@@ -147,9 +150,9 @@ public class ManagedStateImplTest
Slice two = ManagedStateTestUtils.getSliceFor("2");
commitHelper(one, two);
Bucket.DefaultBucket defaultBucket = (Bucket.DefaultBucket)testMeta.managedState.getBucket(0);
- Assert.assertEquals("value of one", one, defaultBucket.getCommittedData().get(one).getValue());
-
- Assert.assertNull("value of two", defaultBucket.getCommittedData().get(two));
+ Assert.assertEquals("value of one", one, defaultBucket.getCommittedData().firstEntry().getValue().get(one)
+ .getValue());
+ Assert.assertNull("value of two", defaultBucket.getCommittedData().firstEntry().getValue().get(two));
testMeta.managedState.teardown();
}
@@ -163,6 +166,45 @@ public class ManagedStateImplTest
Assert.assertEquals("value of one", one, valFuture.get());
}
+ /**
+ * Fix for APEXMALHAR-2042
+ */
+ @Test
+ public void testFreeWindowTransferRaceCondition() throws Exception
+ {
+ testMeta.managedState.setMaxMemorySize(1);
+ testMeta.managedState.setCheckStateSizeInterval(Duration.millis(1L));
+ testMeta.managedState.setCheckStateSizeInterval(Duration.millis(1L));
+ testMeta.managedState.setup(testMeta.operatorContext);
+
+ int numKeys = 300;
+ long lastWindowId = (long)numKeys;
+
+ for (long windowId = 0L; windowId < lastWindowId; windowId++) {
+ testMeta.managedState.beginWindow(windowId);
+ Slice keyVal = ManagedStateTestUtils.getSliceFor(Long.toString(windowId));
+ testMeta.managedState.put(0L, keyVal, keyVal);
+ testMeta.managedState.endWindow();
+ testMeta.managedState.beforeCheckpoint(windowId);
+ testMeta.managedState.checkpointed(windowId);
+ Thread.sleep(1L);
+ }
+
+ testMeta.managedState.committed(lastWindowId - 2L);
+ lastWindowId++;
+
+ testMeta.managedState.beginWindow(lastWindowId);
+
+ for (int key = numKeys - 1; key > 0; key--) {
+ Slice keyVal = ManagedStateTestUtils.getSliceFor(Integer.toString(key));
+ Slice val = testMeta.managedState.getSync(0L, keyVal);
+ Assert.assertNotNull(val);
+ }
+
+ testMeta.managedState.endWindow();
+ testMeta.managedState.teardown();
+ }
+
private void commitHelper(Slice one, Slice two)
{
testMeta.managedState.setup(testMeta.operatorContext);
@@ -180,4 +222,5 @@ public class ManagedStateImplTest
testMeta.managedState.committed(time);
}
+ private static final Logger LOG = LoggerFactory.getLogger(ManagedStateImplTest.class);
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/1d3d8763/library/src/test/java/org/apache/apex/malhar/lib/state/managed/StateTrackerTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/org/apache/apex/malhar/lib/state/managed/StateTrackerTest.java b/library/src/test/java/org/apache/apex/malhar/lib/state/managed/StateTrackerTest.java
index 95defb8..07e141a 100644
--- a/library/src/test/java/org/apache/apex/malhar/lib/state/managed/StateTrackerTest.java
+++ b/library/src/test/java/org/apache/apex/malhar/lib/state/managed/StateTrackerTest.java
@@ -146,9 +146,9 @@ public class StateTrackerTest
}
@Override
- public long freeMemory() throws IOException
+ public long freeMemory(long windowId) throws IOException
{
- long freedBytes = super.freeMemory();
+ long freedBytes = super.freeMemory(windowId);
((MockManagedStateImpl)managedStateContext).freedBuckets.add(getBucketId());
((MockManagedStateImpl)managedStateContext).latch.countDown();
return freedBytes;
[2/2] incubator-apex-malhar git commit: Merge branch
'APEXMALHAR-2042_pull_1'
Posted by cs...@apache.org.
Merge branch 'APEXMALHAR-2042_pull_1'
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/c4052f8c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/c4052f8c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/c4052f8c
Branch: refs/heads/master
Commit: c4052f8c665edaa07ac80b083d8a3be614d5721a
Parents: 4cffeb3 1d3d876
Author: Chandni Singh <cs...@apache.org>
Authored: Mon Apr 11 15:14:19 2016 -0700
Committer: Chandni Singh <cs...@apache.org>
Committed: Mon Apr 11 15:14:19 2016 -0700
----------------------------------------------------------------------
.../apex/malhar/lib/state/managed/Bucket.java | 42 ++++++++++++-----
.../managed/IncrementalCheckpointManager.java | 16 +++++--
.../malhar/lib/state/managed/StateTracker.java | 2 +-
.../lib/state/managed/DefaultBucketTest.java | 2 +-
.../lib/state/managed/ManagedStateImplTest.java | 49 ++++++++++++++++++--
.../lib/state/managed/StateTrackerTest.java | 4 +-
6 files changed, 94 insertions(+), 21 deletions(-)
----------------------------------------------------------------------