Github user Aitozi commented on a diff in the pull request: https://github.com/apache/flink/pull/6205#discussion_r197629205 --- Diff: flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBuffer.java --- @@ -346,16 +351,87 @@ private void lockEvent(EventId eventId) throws Exception { * @throws Exception Thrown if the system cannot access the state. */ public void releaseEvent(EventId eventId) throws Exception { - Lockable<V> eventWrapper = eventsBuffer.get(eventId); + Lockable<V> eventWrapper = getEvent(eventId); if (eventWrapper != null) { if (eventWrapper.release()) { eventsBuffer.remove(eventId); + eventsBufferCache.remove(eventId); } else { - eventsBuffer.put(eventId, eventWrapper); + cacheEvent(eventId, eventWrapper); } } } + // Cache related method + + ///////////////////////////////////////////// + // Put + ///////////////////////////////////////////// + + /** + * Put a event to cache. + * @param eventId id of the event + * @param event event body + */ + private void cacheEvent(EventId eventId, Lockable<V> event) { + this.eventsBufferCache.put(eventId, event); + } + + /** + * Put a ShareBufferNode to cache. + * @param nodeId id of the event + * @param entry SharedBufferNode + */ + private void cacheEntry(NodeId nodeId, Lockable<SharedBufferNode> entry) { + this.entryCache.put(nodeId, entry); + } + + ///////////////////////////////////////////// + // Get + ///////////////////////////////////////////// + + /** + * Try to get the sharedBufferNode from state iff the node has not been quered during this turn process. + * @param nodeId id of the event --- End diff -- it means `if and only if`.
---