You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/02/20 22:10:41 UTC
[GitHub] [flink-statefun] igalshilman opened a new pull request #29:
[FLINK-16063] Add simple backpressure
igalshilman opened a new pull request #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29
# WIP - This PR adds a simple backpressure mechanism.
There are two ways that a backpressure can be triggered:
1. When the number of asynchronous operations per task exceeds a predefined threshold[1]
2. When a prebuilt internal function requires to backpressure a specific address[2]
via calling an internal method on `ReusableContext`.
Please note that, currently if a single function instance requires backpressure, then the result is that all other functions scheduled into the same operator are blocked.
[1] currently that threshold is disabled, and it would be wired in once #27 would be merged.
[2] Although the API allows blocking for a specific address, in the current implementation all the input (for all the addresses) would be blocked.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink-statefun] igalshilman commented on a change in pull request
#29: [FLINK-16063] Add simple backpressure
Posted by GitBox <gi...@apache.org>.
igalshilman commented on a change in pull request #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29#discussion_r382544542
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/backpressure/ThresholdBackPressureValve.java
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.statefun.flink.core.backpressure;
+
+import it.unimi.dsi.fastutil.objects.ObjectOpenHashMap;
+import java.util.Objects;
+import org.apache.flink.statefun.sdk.Address;
+
+/** A simple Threshold based {@link BackPressureValve}. */
+public final class ThresholdBackPressureValve implements BackPressureValve {
+ private final int maximumPendingAsynchronousOperations;
+
+ /**
+ * a set of address that had explicitly requested to stop processing any new inputs (via {@link
+ * AsyncWaiter#awaitAsyncOperationComplete()}. Note that this is a set implemented on top of a
+ * map, and the value (Boolean) has no meaning.
+ */
+ private final ObjectOpenHashMap<Address, Boolean> blockedAddressSet =
+ new ObjectOpenHashMap<>(1024);
+
+ private int pendingAsynchronousOperationsCount;
+
+ /**
+ * Constructs a ThresholdBackPressureValve.
+ *
+ * @param maximumPendingAsynchronousOperations the total allowed async operations to be inflight
+ * per StreamTask, or {@code -1} to disable back pressure.
+ */
+ public ThresholdBackPressureValve(int maximumPendingAsynchronousOperations) {
+ this.maximumPendingAsynchronousOperations = maximumPendingAsynchronousOperations;
+ }
+
+ public boolean shouldBackPressure() {
+ return totalPendingAsyncOperationsAtCapacity() || hasBlockedAddress();
+ }
+
+ public void blockAddress(Address address) {
Review comment:
Hi @tzulitai, your understanding is correct, If a function would request to block without actually registering any async operations then it would remain blocked.
This is an acceptable behaviour since this is our internal API to be used by `HttpFunction` and `GrpcFunction`.
If we would like it to be a part of the SDK then we would have to make sure that we track every async operation registered per each address.
Let me update the JavaDoc of the `AsyncWaiter` with your observation.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink-statefun] tzulitai commented on a change in pull request
#29: [FLINK-16063] Add simple backpressure
Posted by GitBox <gi...@apache.org>.
tzulitai commented on a change in pull request #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29#discussion_r382552029
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/backpressure/ThresholdBackPressureValve.java
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.statefun.flink.core.backpressure;
+
+import it.unimi.dsi.fastutil.objects.ObjectOpenHashMap;
+import java.util.Objects;
+import org.apache.flink.statefun.sdk.Address;
+
+/** A simple Threshold based {@link BackPressureValve}. */
+public final class ThresholdBackPressureValve implements BackPressureValve {
+ private final int maximumPendingAsynchronousOperations;
+
+ /**
+ * a set of address that had explicitly requested to stop processing any new inputs (via {@link
+ * AsyncWaiter#awaitAsyncOperationComplete()}. Note that this is a set implemented on top of a
+ * map, and the value (Boolean) has no meaning.
+ */
+ private final ObjectOpenHashMap<Address, Boolean> blockedAddressSet =
+ new ObjectOpenHashMap<>(1024);
+
+ private int pendingAsynchronousOperationsCount;
+
+ /**
+ * Constructs a ThresholdBackPressureValve.
+ *
+ * @param maximumPendingAsynchronousOperations the total allowed async operations to be inflight
+ * per StreamTask, or {@code -1} to disable back pressure.
+ */
+ public ThresholdBackPressureValve(int maximumPendingAsynchronousOperations) {
+ this.maximumPendingAsynchronousOperations = maximumPendingAsynchronousOperations;
+ }
+
+ public boolean shouldBackPressure() {
+ return totalPendingAsyncOperationsAtCapacity() || hasBlockedAddress();
+ }
+
+ public void blockAddress(Address address) {
Review comment:
I see, understood now thank you!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink-statefun] igalshilman commented on a change in pull request
#29: [FLINK-16063] Add simple backpressure
Posted by GitBox <gi...@apache.org>.
igalshilman commented on a change in pull request #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29#discussion_r382549046
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/backpressure/ThresholdBackPressureValve.java
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.statefun.flink.core.backpressure;
+
+import it.unimi.dsi.fastutil.objects.ObjectOpenHashMap;
+import java.util.Objects;
+import org.apache.flink.statefun.sdk.Address;
+
+/** A simple Threshold based {@link BackPressureValve}. */
+public final class ThresholdBackPressureValve implements BackPressureValve {
+ private final int maximumPendingAsynchronousOperations;
+
+ /**
+ * a set of address that had explicitly requested to stop processing any new inputs (via {@link
+ * AsyncWaiter#awaitAsyncOperationComplete()}. Note that this is a set implemented on top of a
+ * map, and the value (Boolean) has no meaning.
+ */
+ private final ObjectOpenHashMap<Address, Boolean> blockedAddressSet =
+ new ObjectOpenHashMap<>(1024);
+
+ private int pendingAsynchronousOperationsCount;
+
+ /**
+ * Constructs a ThresholdBackPressureValve.
+ *
+ * @param maximumPendingAsynchronousOperations the total allowed async operations to be inflight
+ * per StreamTask, or {@code -1} to disable back pressure.
+ */
+ public ThresholdBackPressureValve(int maximumPendingAsynchronousOperations) {
+ this.maximumPendingAsynchronousOperations = maximumPendingAsynchronousOperations;
+ }
+
+ public boolean shouldBackPressure() {
+ return totalPendingAsyncOperationsAtCapacity() || hasBlockedAddress();
+ }
+
+ public void blockAddress(Address address) {
+ Objects.requireNonNull(address);
+ blockedAddressSet.put(address, Boolean.TRUE);
+ }
+
+ public void notifyAsyncOperationRegistered() {
+ pendingAsynchronousOperationsCount++;
+ }
+
+ public void notifyAsyncOperationCompleted(Address owningAddress) {
+ Objects.requireNonNull(owningAddress);
+ pendingAsynchronousOperationsCount--;
+ blockedAddressSet.remove(owningAddress);
Review comment:
Let me try to clarify:
There are two cases where a backpressure would be triggered:
a) The total number of in-flight async operations in a `StreamTask` exceeds a predefined threshold. This is tracked by `pendingAsynchronousOperationsCount`, it is incremented when an asyncOp is registered, and decremented when it is completed.
b) A specific address has requested to stop processing new inputs, this is tracked by the `blockedAddressSet`.
The method `notifyAsyncOperationCompleted(owningAddress)` is meant to be called when ANY async operation has been completed (with the address supplied as an argument). And the implementation needs to decide what to do with that information.
In the `ThresholdBackPressureValve` we need to account for (a) by decrement the total inflight async operations, and in addition, for (b) we check if that operation can unblock a previously blocked address.
I hope it helps, if it does let me reflect that comment in a JavaDoc of that class.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink-statefun] tzulitai commented on a change in pull request
#29: [FLINK-16063] Add simple backpressure
Posted by GitBox <gi...@apache.org>.
tzulitai commented on a change in pull request #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29#discussion_r382518904
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/backpressure/ThresholdBackPressureValve.java
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.statefun.flink.core.backpressure;
+
+import it.unimi.dsi.fastutil.objects.ObjectOpenHashMap;
+import java.util.Objects;
+import org.apache.flink.statefun.sdk.Address;
+
+/** A simple Threshold based {@link BackPressureValve}. */
+public final class ThresholdBackPressureValve implements BackPressureValve {
+ private final int maximumPendingAsynchronousOperations;
+
+ /**
+ * a set of address that had explicitly requested to stop processing any new inputs (via {@link
+ * AsyncWaiter#awaitAsyncOperationComplete()}. Note that this is a set implemented on top of a
+ * map, and the value (Boolean) has no meaning.
+ */
+ private final ObjectOpenHashMap<Address, Boolean> blockedAddressSet =
+ new ObjectOpenHashMap<>(1024);
+
+ private int pendingAsynchronousOperationsCount;
+
+ /**
+ * Constructs a ThresholdBackPressureValve.
+ *
+ * @param maximumPendingAsynchronousOperations the total allowed async operations to be inflight
+ * per StreamTask, or {@code -1} to disable back pressure.
+ */
+ public ThresholdBackPressureValve(int maximumPendingAsynchronousOperations) {
+ this.maximumPendingAsynchronousOperations = maximumPendingAsynchronousOperations;
+ }
+
+ public boolean shouldBackPressure() {
+ return totalPendingAsyncOperationsAtCapacity() || hasBlockedAddress();
+ }
+
+ public void blockAddress(Address address) {
+ Objects.requireNonNull(address);
+ blockedAddressSet.put(address, Boolean.TRUE);
+ }
+
+ public void notifyAsyncOperationRegistered() {
+ pendingAsynchronousOperationsCount++;
+ }
+
+ public void notifyAsyncOperationCompleted(Address owningAddress) {
+ Objects.requireNonNull(owningAddress);
+ pendingAsynchronousOperationsCount--;
+ blockedAddressSet.remove(owningAddress);
Review comment:
I'm slightly confused here, could you help me a bit with clarifying?:
is `notifyAsyncOperationCompleted` meant to be called when ALL async operations from `owningAddress` has completed, or when SOME async operation from `owningAddress` has completed?
From the `blockedAddressSet.remove(owningAddress)` operation, it seems like its supposed to be the former, but the `pendingAsynchronousOperationsCount` counter keeps track of in-flight asyncs across all addresses, and is not being checked before the address removal.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink-statefun] tzulitai commented on issue #29: [FLINK-16063] Add
simple backpressure
Posted by GitBox <gi...@apache.org>.
tzulitai commented on issue #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29#issuecomment-589725246
+1, LGTM! Thanks for addressing my questions @igalshilman.
I'll wait for one last Travis run before merging.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink-statefun] tzulitai commented on a change in pull request
#29: [FLINK-16063] Add simple backpressure
Posted by GitBox <gi...@apache.org>.
tzulitai commented on a change in pull request #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29#discussion_r382551837
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/backpressure/ThresholdBackPressureValve.java
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.statefun.flink.core.backpressure;
+
+import it.unimi.dsi.fastutil.objects.ObjectOpenHashMap;
+import java.util.Objects;
+import org.apache.flink.statefun.sdk.Address;
+
+/** A simple Threshold based {@link BackPressureValve}. */
+public final class ThresholdBackPressureValve implements BackPressureValve {
+ private final int maximumPendingAsynchronousOperations;
+
+ /**
+ * a set of address that had explicitly requested to stop processing any new inputs (via {@link
+ * AsyncWaiter#awaitAsyncOperationComplete()}. Note that this is a set implemented on top of a
+ * map, and the value (Boolean) has no meaning.
+ */
+ private final ObjectOpenHashMap<Address, Boolean> blockedAddressSet =
+ new ObjectOpenHashMap<>(1024);
+
+ private int pendingAsynchronousOperationsCount;
+
+ /**
+ * Constructs a ThresholdBackPressureValve.
+ *
+ * @param maximumPendingAsynchronousOperations the total allowed async operations to be inflight
+ * per StreamTask, or {@code -1} to disable back pressure.
+ */
+ public ThresholdBackPressureValve(int maximumPendingAsynchronousOperations) {
+ this.maximumPendingAsynchronousOperations = maximumPendingAsynchronousOperations;
+ }
+
+ public boolean shouldBackPressure() {
+ return totalPendingAsyncOperationsAtCapacity() || hasBlockedAddress();
+ }
+
+ public void blockAddress(Address address) {
+ Objects.requireNonNull(address);
+ blockedAddressSet.put(address, Boolean.TRUE);
+ }
+
+ public void notifyAsyncOperationRegistered() {
+ pendingAsynchronousOperationsCount++;
+ }
+
+ public void notifyAsyncOperationCompleted(Address owningAddress) {
+ Objects.requireNonNull(owningAddress);
+ pendingAsynchronousOperationsCount--;
+ blockedAddressSet.remove(owningAddress);
Review comment:
@igalshilman
Thanks, that was very helpful.
Could you add to the Javadoc of `blockAddress` that it is unblocked as soon as some (one) async operation for the address completes?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink-statefun] tzulitai commented on a change in pull request
#29: [FLINK-16063] Add simple backpressure
Posted by GitBox <gi...@apache.org>.
tzulitai commented on a change in pull request #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29#discussion_r382516057
##########
File path: statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/backpressure/ThresholdBackPressureValve.java
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.statefun.flink.core.backpressure;
+
+import it.unimi.dsi.fastutil.objects.ObjectOpenHashMap;
+import java.util.Objects;
+import org.apache.flink.statefun.sdk.Address;
+
+/** A simple Threshold based {@link BackPressureValve}. */
+public final class ThresholdBackPressureValve implements BackPressureValve {
+ private final int maximumPendingAsynchronousOperations;
+
+ /**
+ * a set of address that had explicitly requested to stop processing any new inputs (via {@link
+ * AsyncWaiter#awaitAsyncOperationComplete()}. Note that this is a set implemented on top of a
+ * map, and the value (Boolean) has no meaning.
+ */
+ private final ObjectOpenHashMap<Address, Boolean> blockedAddressSet =
+ new ObjectOpenHashMap<>(1024);
+
+ private int pendingAsynchronousOperationsCount;
+
+ /**
+ * Constructs a ThresholdBackPressureValve.
+ *
+ * @param maximumPendingAsynchronousOperations the total allowed async operations to be inflight
+ * per StreamTask, or {@code -1} to disable back pressure.
+ */
+ public ThresholdBackPressureValve(int maximumPendingAsynchronousOperations) {
+ this.maximumPendingAsynchronousOperations = maximumPendingAsynchronousOperations;
+ }
+
+ public boolean shouldBackPressure() {
+ return totalPendingAsyncOperationsAtCapacity() || hasBlockedAddress();
+ }
+
+ public void blockAddress(Address address) {
Review comment:
What happens if a function blocks itself (via `AsyncWaiter#awaitAsyncOperationComplete()`), when there actually are no in-flight async operations for it?
Since in that case, the `notifyAsyncOperationCompleted` seems to never be invoked for that address, and therefore the blocked address will never be removed from `blockedAddresSet` ==> `hasBlockedAddress` will end up always being false?
Not sure if I'm missing something here, please correct me if I'm wrong!
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink-statefun] tzulitai closed pull request #29: [FLINK-16063]
Add simple backpressure
Posted by GitBox <gi...@apache.org>.
tzulitai closed pull request #29: [FLINK-16063] Add simple backpressure
URL: https://github.com/apache/flink-statefun/pull/29
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services