masteryhx commented on code in PR #24633:
URL: https://github.com/apache/flink/pull/24633#discussion_r1562207787
##########
flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java:
##########
@@ -152,66 +148,168 @@ void testBasicRun() {
// Single-step run for another key.
// Firstly, the user code generates value get in active buffer.
- assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.stateRequestsBuffer.activeQueueSize()).isEqualTo(1);
assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ assertThat(aec.inFlightRecordNum.get()).isEqualTo(1);
aec.triggerIfNeeded(true);
// After running, the value update is in active buffer.
- assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.stateRequestsBuffer.activeQueueSize()).isEqualTo(1);
assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ assertThat(aec.inFlightRecordNum.get()).isEqualTo(1);
aec.triggerIfNeeded(true);
// Value update finishes.
- assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.stateRequestsBuffer.activeQueueSize()).isEqualTo(0);
assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(aec.inFlightRecordNum.get()).isEqualTo(0);
assertThat(output.get()).isEqualTo(1);
assertThat(recordContext4.getReferenceCount()).isEqualTo(0);
}
- /**
- * An AsyncExecutionController for testing purpose, which integrates with
basic buffer
- * mechanism.
- */
- static class TestAsyncExecutionController<R, K> extends
AsyncExecutionController<R, K> {
+ @Test
+ void testRecordsRunInOrder() {
+ AsyncExecutionController<String, String> aec =
+ new AsyncExecutionController<>(new SyncMailboxExecutor(), new
TestStateExecutor());
+ TestUnderlyingState underlyingState = new TestUnderlyingState();
+ TestValueState valueState = new TestValueState(aec, underlyingState);
+ AtomicInteger output = new AtomicInteger();
+ Runnable userCode =
Review Comment:
The logic seems same as other tests, could this be a seprate method ?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.asyncprocessing;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A buffer to hold state requests to execute state requests in batch, which
can only be manipulated
+ * within task thread.
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+@NotThreadSafe
+public class StateRequestBuffer<R, K> {
+ /**
+ * The state requests in this buffer could be executed when the buffer is
full or configured
+ * batch size is reached. All operations on this buffer must be invoked in
task thread.
+ */
+ final LinkedList<StateRequest<K, ?, ?>> activeQueue;
+
+ /**
+ * The requests in that should wait until all preceding records with
identical key finishing its
+ * execution. After which the queueing requests will move into the active
buffer. All operations
+ * on this buffer must be invoked in task thread.
+ */
+ final Map<K, Deque<StateRequest<K, ?, ?>>> blockingQueue;
+
+ /** The number of state requests in blocking queue. */
+ int blockingQueueSize;
+
+ public StateRequestBuffer() {
+ this.activeQueue = new LinkedList<>();
+ this.blockingQueue = new HashMap<>();
+ this.blockingQueueSize = 0;
+ }
+
+ void enqueueToActive(StateRequest<K, ?, ?> request) {
+ activeQueue.add(request);
+ }
+
+ void enqueueToBlocking(StateRequest<K, ?, ?> request) {
+ blockingQueue
+ .computeIfAbsent(request.getRecordContext().getKey(), k -> new
LinkedList<>())
+ .add(request);
+ blockingQueueSize++;
+ }
+
+ /**
+ * Try to pull one state request with specific key from blocking queue to
active queue.
+ *
+ * @param key The key to release, the other records with this key is no
longer blocking.
+ * @return The first record context with the same key in blocking queue,
null if no such record.
+ */
+ @Nullable
+ RecordContext<R, K> tryActivateOneByKey(K key) {
+ if (!blockingQueue.containsKey(key)) {
+ return null;
+ }
+
+ StateRequest<K, ?, ?> stateRequest = blockingQueue.get(key).getFirst();
Review Comment:
How about just using `removeFirst()` and get `StateRequest` here ?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequestBuffer.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.asyncprocessing;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A buffer to hold state requests to execute state requests in batch, which
can only be manipulated
+ * within task thread.
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+@NotThreadSafe
+public class StateRequestBuffer<R, K> {
+ /**
+ * The state requests in this buffer could be executed when the buffer is
full or configured
+ * batch size is reached. All operations on this buffer must be invoked in
task thread.
+ */
+ final LinkedList<StateRequest<K, ?, ?>> activeQueue;
+
+ /**
+ * The requests in that should wait until all preceding records with
identical key finishing its
+ * execution. After which the queueing requests will move into the active
buffer. All operations
+ * on this buffer must be invoked in task thread.
+ */
+ final Map<K, Deque<StateRequest<K, ?, ?>>> blockingQueue;
+
+ /** The number of state requests in blocking queue. */
+ int blockingQueueSize;
+
+ public StateRequestBuffer() {
+ this.activeQueue = new LinkedList<>();
+ this.blockingQueue = new HashMap<>();
+ this.blockingQueueSize = 0;
+ }
+
+ void enqueueToActive(StateRequest<K, ?, ?> request) {
+ activeQueue.add(request);
+ }
+
+ void enqueueToBlocking(StateRequest<K, ?, ?> request) {
+ blockingQueue
+ .computeIfAbsent(request.getRecordContext().getKey(), k -> new
LinkedList<>())
+ .add(request);
+ blockingQueueSize++;
+ }
+
+ /**
+ * Try to pull one state request with specific key from blocking queue to
active queue.
+ *
+ * @param key The key to release, the other records with this key is no
longer blocking.
+ * @return The first record context with the same key in blocking queue,
null if no such record.
+ */
+ @Nullable
+ RecordContext<R, K> tryActivateOneByKey(K key) {
+ if (!blockingQueue.containsKey(key)) {
+ return null;
+ }
+
+ StateRequest<K, ?, ?> stateRequest = blockingQueue.get(key).getFirst();
+ activeQueue.add(stateRequest);
+ blockingQueue.get(key).removeFirst();
+ if (blockingQueue.get(key).isEmpty()) {
+ blockingQueue.remove(key);
+ }
+ blockingQueueSize--;
+ return (RecordContext<R, K>) stateRequest.getRecordContext();
Review Comment:
minor suggestion: add `SuppressWarnings` for this ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
