You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Fanoid (via GitHub)" <gi...@apache.org> on 2023/08/08 12:05:24 UTC
[GitHub] [flink-ml] Fanoid opened a new pull request, #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Fanoid opened a new pull request, #250:
URL: https://github.com/apache/flink-ml/pull/250
## What is the purpose of the change
Right now, by default, an instance of `ListStateWithCache` uses up all the managed memory allocated for `ManagedMemoryUseCase.OPERATOR`.
It could bring some bugs in some situations, for example, when there exist more than one `ListStateWithCache`s in a single operator, or there are other places using managed memory of `ManagedMemoryUseCase.OPERATOR`.
This PR tries to fix bugs with such default behaviors.
## Brief change log
- Adds a parameter `memorySubFraction` to the constructor of `ListStateWithCache`, so that the developer of the operator is aware about the usage of managed memory, and can control its usage.
- Changes the default behavior of `ListStateWithCache` to use none of managed memory instead of use all.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? JavaDocs
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] Fanoid commented on pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on PR #250:
URL: https://github.com/apache/flink-ml/pull/250#issuecomment-1693027114
@jiangxin369 Thanks. Please check the changes and above responses.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] Fanoid commented on pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on PR #250:
URL: https://github.com/apache/flink-ml/pull/250#issuecomment-1685708251
hi, @jiangxin369 , would you take another look of latest changes?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] Fanoid commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1294247032
##########
flink-ml-core/src/test/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCacheTest.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * 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.iteration.datacache.nonkeyed;
+
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.memory.MemoryReservationException;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
+
+/** Tests {@link ListStateWithCache}. */
+public class ListStateWithCacheTest {
+
+ @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ private MiniClusterConfiguration createMiniClusterConfiguration() {
+ Configuration configuration = new Configuration();
+ // Set managed memory size to a small value, so when the instance of ListStateWithCache
+ // tries to allocate memory, it will use up all managed memory assigned to itself.
+ configuration.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.ofMebiBytes(16));
+ configuration.set(RestOptions.PORT, 18082);
Review Comment:
I've removed this config. Thanks.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] Fanoid commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1296795610
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -62,14 +63,32 @@ public class ListStateWithCache<T> implements ListState<T> {
/** The data cache writer for the received records. */
private final DataCacheWriter<T> dataCacheWriter;
- @SuppressWarnings("unchecked")
public ListStateWithCache(
TypeSerializer<T> serializer,
StreamTask<?, ?> containingTask,
StreamingRuntimeContext runtimeContext,
StateInitializationContext stateInitializationContext,
OperatorID operatorID)
throws IOException {
+ this(
+ serializer,
+ containingTask,
+ runtimeContext,
+ stateInitializationContext,
+ operatorID,
+ 0.);
Review Comment:
@jiangxin369 and @zhipeng93 , thanks for your comments.
After offline discussion, I found it is necessary to do check about sub-fractions from all usages on the operator-scope managed memory. So I changed the constructor of `ListStateWithCache`.
Please check the latest changes.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] Fanoid commented on pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on PR #250:
URL: https://github.com/apache/flink-ml/pull/250#issuecomment-1681787714
hi, @jiangxin369 and @zhipeng93, I've updated the PR. Could you have another look?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "zhipeng93 (via GitHub)" <gi...@apache.org>.
zhipeng93 commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1294355543
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -62,14 +63,32 @@ public class ListStateWithCache<T> implements ListState<T> {
/** The data cache writer for the received records. */
private final DataCacheWriter<T> dataCacheWriter;
- @SuppressWarnings("unchecked")
public ListStateWithCache(
TypeSerializer<T> serializer,
StreamTask<?, ?> containingTask,
StreamingRuntimeContext runtimeContext,
StateInitializationContext stateInitializationContext,
OperatorID operatorID)
throws IOException {
+ this(
+ serializer,
+ containingTask,
+ runtimeContext,
+ stateInitializationContext,
+ operatorID,
+ 0.);
Review Comment:
Would it be better to set the default value as 1.0?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1303865404
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -18,33 +18,75 @@
package org.apache.flink.iteration.datacache.nonkeyed;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.util.Preconditions;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.WeakHashMap;
/**
* A manager for operator-scope managed memory.
*
- * <p>Every operator must create one (and only one) instance of this class to manage usages of
- * operator-scope managed memery. In the operator, for every usage, {@link #register} is called to
- * declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
- * calling {@link #getFraction}.
+ * <p>Every operator must call {@link #getOrCreate} to get an instance to manage usages of
+ * operator-scope managed memory.
*
- * <p>Note that all calls of {@link #register} must be before those of {@link #getFraction}.
+ * <p>In the operator, for every usage of operator-scope managed memory, {@link #register} is called
+ * to declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
+ * calling {@link #getFraction}. Note that all calls of {@link #register} must be before those of
+ * {@link #getFraction}.
*/
+@PublicEvolving
public class OperatorScopeManagedMemoryManager {
+ /**
+ * Stores instances corresponding to operators. The instance is expected to be released after
+ * some point after the corresponding operator ID is unused.
+ */
+ private static final Map<OperatorID, OperatorScopeManagedMemoryManager> managers =
+ Collections.synchronizedMap(new WeakHashMap<>());
+
+ /** Stores keys and weights of memory usages. */
protected Map<String, Double> weights = new HashMap<>();
+ /** Indicates whether the `weights` is frozen. */
protected boolean frozen = false;
+ /** Stores sum of weights of all usages. */
protected double sum;
+ OperatorScopeManagedMemoryManager() {}
Review Comment:
Can we remove this empty constructor?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1303865404
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -18,33 +18,75 @@
package org.apache.flink.iteration.datacache.nonkeyed;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.util.Preconditions;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.WeakHashMap;
/**
* A manager for operator-scope managed memory.
*
- * <p>Every operator must create one (and only one) instance of this class to manage usages of
- * operator-scope managed memery. In the operator, for every usage, {@link #register} is called to
- * declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
- * calling {@link #getFraction}.
+ * <p>Every operator must call {@link #getOrCreate} to get an instance to manage usages of
+ * operator-scope managed memory.
*
- * <p>Note that all calls of {@link #register} must be before those of {@link #getFraction}.
+ * <p>In the operator, for every usage of operator-scope managed memory, {@link #register} is called
+ * to declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
+ * calling {@link #getFraction}. Note that all calls of {@link #register} must be before those of
+ * {@link #getFraction}.
*/
+@PublicEvolving
public class OperatorScopeManagedMemoryManager {
+ /**
+ * Stores instances corresponding to operators. The instance is expected to be released after
+ * some point after the corresponding operator ID is unused.
+ */
+ private static final Map<OperatorID, OperatorScopeManagedMemoryManager> managers =
+ Collections.synchronizedMap(new WeakHashMap<>());
+
+ /** Stores keys and weights of memory usages. */
protected Map<String, Double> weights = new HashMap<>();
+ /** Indicates whether the `weights` is frozen. */
protected boolean frozen = false;
+ /** Stores sum of weights of all usages. */
protected double sum;
+ OperatorScopeManagedMemoryManager() {}
Review Comment:
Can we remove this empty constructor?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] Fanoid commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1305396094
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -18,33 +18,75 @@
package org.apache.flink.iteration.datacache.nonkeyed;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.util.Preconditions;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.WeakHashMap;
/**
* A manager for operator-scope managed memory.
*
- * <p>Every operator must create one (and only one) instance of this class to manage usages of
- * operator-scope managed memery. In the operator, for every usage, {@link #register} is called to
- * declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
- * calling {@link #getFraction}.
+ * <p>Every operator must call {@link #getOrCreate} to get an instance to manage usages of
+ * operator-scope managed memory.
*
- * <p>Note that all calls of {@link #register} must be before those of {@link #getFraction}.
+ * <p>In the operator, for every usage of operator-scope managed memory, {@link #register} is called
+ * to declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
+ * calling {@link #getFraction}. Note that all calls of {@link #register} must be before those of
+ * {@link #getFraction}.
*/
+@PublicEvolving
public class OperatorScopeManagedMemoryManager {
+ /**
+ * Stores instances corresponding to operators. The instance is expected to be released after
+ * some point after the corresponding operator ID is unused.
+ */
+ private static final Map<OperatorID, OperatorScopeManagedMemoryManager> managers =
+ Collections.synchronizedMap(new WeakHashMap<>());
+
+ /** Stores keys and weights of memory usages. */
protected Map<String, Double> weights = new HashMap<>();
+ /** Indicates whether the `weights` is frozen. */
protected boolean frozen = false;
+ /** Stores sum of weights of all usages. */
protected double sum;
+ OperatorScopeManagedMemoryManager() {}
Review Comment:
The empty constructor is actually used to limit its access.
But I think using `private` scope is more reasonable here.
Please check the changes.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1303885405
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -95,13 +92,18 @@ public ListStateWithCache(
ManagedMemoryUseCase.OPERATOR,
runtimeContext.getTaskManagerRuntimeInfo().getConfiguration(),
runtimeContext.getUserCodeClassLoader());
- if (fraction * memorySubFraction > 0) {
- MemoryManager memoryManager = containingTask.getEnvironment().getMemoryManager();
- segmentPool =
- new LazyMemorySegmentPool(
- containingTask,
- memoryManager,
- memoryManager.computeNumberOfPages(fraction * memorySubFraction));
+ if (fraction > 0) {
+ OperatorScopeManagedMemoryManager manager =
+ OperatorScopeManagedMemoryManager.getOrCreate(operatorID);
+ double memorySubFraction = manager.getFraction(key);
+ if (fraction * memorySubFraction > 0) {
Review Comment:
`if (memorySubFraction > 0)` is enough.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -18,33 +18,75 @@
package org.apache.flink.iteration.datacache.nonkeyed;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.util.Preconditions;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.WeakHashMap;
/**
* A manager for operator-scope managed memory.
*
- * <p>Every operator must create one (and only one) instance of this class to manage usages of
- * operator-scope managed memery. In the operator, for every usage, {@link #register} is called to
- * declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
- * calling {@link #getFraction}.
+ * <p>Every operator must call {@link #getOrCreate} to get an instance to manage usages of
+ * operator-scope managed memory.
*
- * <p>Note that all calls of {@link #register} must be before those of {@link #getFraction}.
+ * <p>In the operator, for every usage of operator-scope managed memory, {@link #register} is called
+ * to declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
+ * calling {@link #getFraction}. Note that all calls of {@link #register} must be before those of
+ * {@link #getFraction}.
*/
+@PublicEvolving
public class OperatorScopeManagedMemoryManager {
+ /**
+ * Stores instances corresponding to operators. The instance is expected to be released after
+ * some point after the corresponding operator ID is unused.
+ */
+ private static final Map<OperatorID, OperatorScopeManagedMemoryManager> managers =
+ Collections.synchronizedMap(new WeakHashMap<>());
+
+ /** Stores keys and weights of memory usages. */
protected Map<String, Double> weights = new HashMap<>();
+ /** Indicates whether the `weights` is frozen. */
protected boolean frozen = false;
+ /** Stores sum of weights of all usages. */
protected double sum;
+ OperatorScopeManagedMemoryManager() {}
Review Comment:
Can we remove this empty constructor?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on PR #250:
URL: https://github.com/apache/flink-ml/pull/250#issuecomment-1700416201
LGTM.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 commented on pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "zhipeng93 (via GitHub)" <gi...@apache.org>.
zhipeng93 commented on PR #250:
URL: https://github.com/apache/flink-ml/pull/250#issuecomment-1702275129
Thanks @Fanoid for the fix and @jiangxin369 for the review. LGTM.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1292700690
##########
flink-ml-core/src/test/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCacheTest.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * 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.iteration.datacache.nonkeyed;
+
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.MemorySize;
+import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.configuration.TaskManagerOptions;
+import org.apache.flink.ml.common.datastream.DataStreamUtils;
+import org.apache.flink.ml.util.TestUtils;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.memory.MemoryReservationException;
+import org.apache.flink.runtime.minicluster.MiniCluster;
+import org.apache.flink.runtime.minicluster.MiniClusterConfiguration;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorStateHandler;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
+
+/** Tests {@link ListStateWithCache}. */
+public class ListStateWithCacheTest {
+
+ @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ private MiniClusterConfiguration createMiniClusterConfiguration() {
+ Configuration configuration = new Configuration();
+ // Set managed memory size to a small value, so when the instance of ListStateWithCache
+ // tries to allocate memory, it will use up all managed memory assigned to itself.
+ configuration.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.ofMebiBytes(16));
+ configuration.set(RestOptions.PORT, 18082);
Review Comment:
Do we need this config?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -62,14 +63,32 @@ public class ListStateWithCache<T> implements ListState<T> {
/** The data cache writer for the received records. */
private final DataCacheWriter<T> dataCacheWriter;
- @SuppressWarnings("unchecked")
public ListStateWithCache(
TypeSerializer<T> serializer,
StreamTask<?, ?> containingTask,
StreamingRuntimeContext runtimeContext,
StateInitializationContext stateInitializationContext,
OperatorID operatorID)
throws IOException {
+ this(
+ serializer,
+ containingTask,
+ runtimeContext,
+ stateInitializationContext,
+ operatorID,
+ 0.);
Review Comment:
With this constructor, the current algorithms using `ListStateWithCache` like `Kmeans` would not use managed memory at all, which causes performance regression.
Would it be better if we divide the memory evenly if multiple `ListStateWithCache` is used by default?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1292699290
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -62,14 +63,32 @@ public class ListStateWithCache<T> implements ListState<T> {
/** The data cache writer for the received records. */
private final DataCacheWriter<T> dataCacheWriter;
- @SuppressWarnings("unchecked")
public ListStateWithCache(
TypeSerializer<T> serializer,
StreamTask<?, ?> containingTask,
StreamingRuntimeContext runtimeContext,
StateInitializationContext stateInitializationContext,
OperatorID operatorID)
throws IOException {
+ this(
+ serializer,
+ containingTask,
+ runtimeContext,
+ stateInitializationContext,
+ operatorID,
+ 0.);
Review Comment:
With this constructor, the current algorithms using `ListStateWithCache` like `Kmeans` would not use managed memory at all, which causes performance regression.
Would it be better if we divide the memory evenly if multiple `ListStateWithCache` is declared by default?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1292699290
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -62,14 +63,32 @@ public class ListStateWithCache<T> implements ListState<T> {
/** The data cache writer for the received records. */
private final DataCacheWriter<T> dataCacheWriter;
- @SuppressWarnings("unchecked")
public ListStateWithCache(
TypeSerializer<T> serializer,
StreamTask<?, ?> containingTask,
StreamingRuntimeContext runtimeContext,
StateInitializationContext stateInitializationContext,
OperatorID operatorID)
throws IOException {
+ this(
+ serializer,
+ containingTask,
+ runtimeContext,
+ stateInitializationContext,
+ operatorID,
+ 0.);
Review Comment:
With this constructor, the current algorithms using `ListStateWithCache` like `Kmeans` would not use managed memory at all, which causes performance regression.
Would it be better if we divide the memory evenly if multiple `ListStateWithCache` is declared by default?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] Fanoid commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1305396094
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -18,33 +18,75 @@
package org.apache.flink.iteration.datacache.nonkeyed;
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.util.Preconditions;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import java.util.WeakHashMap;
/**
* A manager for operator-scope managed memory.
*
- * <p>Every operator must create one (and only one) instance of this class to manage usages of
- * operator-scope managed memery. In the operator, for every usage, {@link #register} is called to
- * declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
- * calling {@link #getFraction}.
+ * <p>Every operator must call {@link #getOrCreate} to get an instance to manage usages of
+ * operator-scope managed memory.
*
- * <p>Note that all calls of {@link #register} must be before those of {@link #getFraction}.
+ * <p>In the operator, for every usage of operator-scope managed memory, {@link #register} is called
+ * to declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
+ * calling {@link #getFraction}. Note that all calls of {@link #register} must be before those of
+ * {@link #getFraction}.
*/
+@PublicEvolving
public class OperatorScopeManagedMemoryManager {
+ /**
+ * Stores instances corresponding to operators. The instance is expected to be released after
+ * some point after the corresponding operator ID is unused.
+ */
+ private static final Map<OperatorID, OperatorScopeManagedMemoryManager> managers =
+ Collections.synchronizedMap(new WeakHashMap<>());
+
+ /** Stores keys and weights of memory usages. */
protected Map<String, Double> weights = new HashMap<>();
+ /** Indicates whether the `weights` is frozen. */
protected boolean frozen = false;
+ /** Stores sum of weights of all usages. */
protected double sum;
+ OperatorScopeManagedMemoryManager() {}
Review Comment:
The empty constructor is actually used to limit its access. But I think using `private` scope is more reasonable here.
Please check the changes.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] zhipeng93 merged pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "zhipeng93 (via GitHub)" <gi...@apache.org>.
zhipeng93 merged PR #250:
URL: https://github.com/apache/flink-ml/pull/250
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] jiangxin369 commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1298134912
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.iteration.datacache.nonkeyed;
+
+import org.apache.flink.util.Preconditions;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A manager for operator-scope managed memory.
+ *
+ * <p>Every operator must create one (and only one) instance of this class to manage usages of
Review Comment:
Instead of allowing users to create `OperatorScopeManagedMemoryManager`, would it be better to provide an `OperatorScopeManagedMemoryManagerBroker` which is `@PublicEvolving` to manage `OperatorScopeManagedMemoryManager` of all operators. It helps ensure that only one instance exists in each operator like below.
```
@PublicEvolving
class OperatorScopeManagedMemoryManagerBroker {
private Map<OperatorID, OperatorScopeManagedMemoryManager> managers;
public static OperatorScopeManagedMemoryManager getOrCreate(OperatorID id) {}
}
```
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.iteration.datacache.nonkeyed;
+
+import org.apache.flink.util.Preconditions;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A manager for operator-scope managed memory.
+ *
+ * <p>Every operator must create one (and only one) instance of this class to manage usages of
+ * operator-scope managed memery. In the operator, for every usage, {@link #register} is called to
Review Comment:
```suggestion
* operator-scope managed memory. In the operator, for every usage, {@link #register} is called to
```
##########
flink-ml-lib/src/main/java/org/apache/flink/ml/clustering/kmeans/KMeans.java:
##########
@@ -239,9 +240,13 @@ public void initializeState(StateInitializationContext context) throws Exception
context.getOperatorStateStore()
.getListState(new ListStateDescriptor<>("centroids", type));
+ OperatorScopeManagedMemoryManager manager = new OperatorScopeManagedMemoryManager();
+ manager.register("points-state", 1.);
points =
new ListStateWithCache<>(
new VectorWithNormSerializer(),
+ manager,
+ "points-state",
Review Comment:
It is recommended to declare the reused Strings as a static final constant.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.iteration.datacache.nonkeyed;
+
+import org.apache.flink.util.Preconditions;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A manager for operator-scope managed memory.
+ *
+ * <p>Every operator must create one (and only one) instance of this class to manage usages of
+ * operator-scope managed memery. In the operator, for every usage, {@link #register} is called to
+ * declare its weight of memory usage with a key. Then, the fraction of memory can be obtained by
+ * calling {@link #getFraction}.
+ *
+ * <p>Note that all calls of {@link #register} must be before those of {@link #getFraction}.
+ */
+public class OperatorScopeManagedMemoryManager {
+
+ protected Map<String, Double> weights = new HashMap<>();
+ protected boolean frozen = false;
+ protected double sum;
+
+ public void register(String key, double weight) {
+ Preconditions.checkState(!frozen, "Cannot call register after getFraction is called.");
+ Preconditions.checkArgument(weight >= 0, "The weight must be non-negative.");
+ weights.put(key, weight);
Review Comment:
Would it be better to throw exceptions when the duplicate key is registered, instead of overwrite it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-ml] Fanoid commented on a diff in pull request #250: [FLINK-32810] Improve memory allocation in ListStateWithCache
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on code in PR #250:
URL: https://github.com/apache/flink-ml/pull/250#discussion_r1299581271
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/OperatorScopeManagedMemoryManager.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * 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.iteration.datacache.nonkeyed;
+
+import org.apache.flink.util.Preconditions;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * A manager for operator-scope managed memory.
+ *
+ * <p>Every operator must create one (and only one) instance of this class to manage usages of
Review Comment:
Good point! I've updated the changes by adding `getOrCreate` in the `OperatorScopeManagedMemoryManager`. Please check it.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org