[FLINK-9642][cep] Added caching layer to SharedBuffer(apache#6205)

sceee · Aug 29, 2018 · 260f789 · 260f789
1 parent 7eb41f3
commit 260f789
Show file tree

Hide file tree

Showing 11 changed files with 784 additions and 523 deletions.
diff --git a/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java b/flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java
diff --git a/...s/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/AfterMatchSkipStrategy.java b/...s/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/AfterMatchSkipStrategy.java
@@ -20,7 +20,7 @@
 
 import org.apache.flink.cep.nfa.ComputationState;
 import org.apache.flink.cep.nfa.sharedbuffer.EventId;
-import org.apache.flink.cep.nfa.sharedbuffer.SharedBuffer;
+import org.apache.flink.cep.nfa.sharedbuffer.SharedBufferAccessor;
 
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -87,21 +87,21 @@ public static AfterMatchSkipStrategy noSkip() {
 	 *
 	 * @param matchesToPrune current partial matches
 	 * @param matchedResult  already completed matches
-	 * @param sharedBuffer   corresponding shared buffer
+	 * @param sharedBufferAccessor   accessor to corresponding shared buffer
 	 * @throws Exception thrown if could not access the state
 	 */
 	public void prune(
 			Collection<ComputationState> matchesToPrune,
 			Collection<Map<String, List<EventId>>> matchedResult,
-			SharedBuffer<?> sharedBuffer) throws Exception {
+			SharedBufferAccessor<?> sharedBufferAccessor) throws Exception {
 
 		EventId pruningId = getPruningId(matchedResult);
 		if (pruningId != null) {
 			List<ComputationState> discardStates = new ArrayList<>();
 			for (ComputationState computationState : matchesToPrune) {
 				if (computationState.getStartEventID() != null &&
 					shouldPrune(computationState.getStartEventID(), pruningId)) {
-					sharedBuffer.releaseNode(computationState.getPreviousBufferEntry());
+					sharedBufferAccessor.releaseNode(computationState.getPreviousBufferEntry());
 					discardStates.add(computationState);
 				}
 			}