You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "jiangxin369 (via GitHub)" <gi...@apache.org> on 2023/07/31 01:42:42 UTC
[GitHub] [flink-ml] jiangxin369 opened a new pull request, #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
jiangxin369 opened a new pull request, #248:
URL: https://github.com/apache/flink-ml/pull/248
<!--
*Thank you very much for contributing to Apache Flink ML - we are happy that you want to help us improve Flink ML. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to one [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] Title of the pull request`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Supports spilling to disk when feedback channel memory buffer is full.
## Brief change log
- Adds config to allow user to set the in-memory buffer size.
- Adds SpillableFeedbackChannel to support spilling to disk automatically.
## 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)`: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- 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] jiangxin369 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1298092153
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/HeadOperator.java:
##########
@@ -422,16 +430,44 @@ public void close() throws Exception {
}
}
- private void registerFeedbackConsumer(Executor mailboxExecutor) {
+ private void registerFeedbackConsumer(Executor mailboxExecutor)
+ throws MemoryAllocationException {
+ StreamTask<?, ?> task = getContainingTask();
+
int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
int attemptNum = getRuntimeContext().getAttemptNumber();
FeedbackKey<StreamRecord<IterationRecord<?>>> feedbackKey =
OperatorUtils.createFeedbackKey(iterationId, feedbackIndex);
SubtaskFeedbackKey<StreamRecord<IterationRecord<?>>> key =
feedbackKey.withSubTaskIndex(indexOfThisSubtask, attemptNum);
- FeedbackChannelBroker broker = FeedbackChannelBroker.get();
- FeedbackChannel<StreamRecord<IterationRecord<?>>> channel = broker.getChannel(key);
- OperatorUtils.registerFeedbackConsumer(channel, this, mailboxExecutor);
+ SpillableFeedbackChannelBroker broker = SpillableFeedbackChannelBroker.get();
+ this.feedbackChannel = broker.getChannel(key);
Review Comment:
I agree with your 2nd point and I will pass the initialization logic as a `Consumer`.
But the HeadOperator and TailOperator still need to coordinate on accessing the shared channel. Because only the operator that called `declareManagedMemoryUseCaseAtOperatorScope` can initialize the Channel, while the other cannot since it has no managed memory to allocate.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1297236349
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
Review Comment:
`SpillingResettableIterator` is not similar to our requirement because it just iterates an existing collection and doesn't need to re-allocate more memories, however, the `SpillableFeedbackQueue` needs to receive, iterate and free items again and again. With the given API of `SpillingBuffer`, I think we need to release and re-allocate the buffers.
Btw, I think the allocation is just taking out a piece of memories from the reserved ones which doesn't cost much.
--
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] lindong28 merged pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 merged PR #248:
URL: https://github.com/apache/flink-ml/pull/248
--
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] lindong28 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1298117222
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/typeinfo/IterationRecordSerializer.java:
##########
@@ -106,7 +106,7 @@ public int getLength() {
@Override
public void serialize(IterationRecord<T> record, DataOutputView target) throws IOException {
target.writeByte((byte) record.getType().ordinal());
- serializerNumber(record.getEpoch(), target);
+ target.writeInt(record.getEpoch());
Review Comment:
How about we create something like `public final END_EPOCH_WATERMARK = Integer.MAX_VALUE - 1 ` and use this variable in e.g. `TerminatingHeadOperatorRecordProcessor`? Then the code can be readable without increasing metadata overhead.
--
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] lindong28 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1299531846
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
Review Comment:
Just to summarize our offline discussions:
- Currently MemoryManager#allocatePages will allocate memory from JVM rather than re-using allocated memory from a pool. We should avoid this repeated memory allocation.
- AbstractPagedOutputView, the parent class of SpillingBuffer, provides clear() and advance() to support writing to a buffer after it has been read. It might be possible to let `SpillingBuffer` implement these two APIs so that we can re-use it for writing after it is read.
- NormalizedKeySorter#reset supports re-using a buffer after it is read.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1297251801
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/HeadOperator.java:
##########
@@ -422,16 +430,44 @@ public void close() throws Exception {
}
}
- private void registerFeedbackConsumer(Executor mailboxExecutor) {
+ private void registerFeedbackConsumer(Executor mailboxExecutor)
+ throws MemoryAllocationException {
+ StreamTask<?, ?> task = getContainingTask();
+
int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
int attemptNum = getRuntimeContext().getAttemptNumber();
FeedbackKey<StreamRecord<IterationRecord<?>>> feedbackKey =
OperatorUtils.createFeedbackKey(iterationId, feedbackIndex);
SubtaskFeedbackKey<StreamRecord<IterationRecord<?>>> key =
feedbackKey.withSubTaskIndex(indexOfThisSubtask, attemptNum);
- FeedbackChannelBroker broker = FeedbackChannelBroker.get();
- FeedbackChannel<StreamRecord<IterationRecord<?>>> channel = broker.getChannel(key);
- OperatorUtils.registerFeedbackConsumer(channel, this, mailboxExecutor);
+ SpillableFeedbackChannelBroker broker = SpillableFeedbackChannelBroker.get();
+ this.feedbackChannel = broker.getChannel(key);
Review Comment:
Sorry, I don's see the benefit of putting the logic in a Consumer, am I misunderstood? IMO, the current implementation of getChannal(key) / removeChannel(key) seems symmetrical and readable.
--
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] lindong28 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1299531846
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
Review Comment:
Just to summarize our offline discussions:
- Currently MemoryManager#allocatePages will allocate memory from JVM rather than re-using allocated memory from a pool. We should avoid this repeated memory allocation.
- AbstractPagedOutputView, the parent class of SpillingBuffer, provides clear() and advance() to support writing to a buffer after it has been read. It might be possible to let `SpillingBuffer` implement these two APIs so that we can re-use it for writing after it is read.
- NormalizedKeySorter#reset supports re-using a buffer after it is read.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1280468831
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+public class SpillableQueue<T> {
Review Comment:
Maybe name it `SpillableFeedbackQueue` is more suitable, as the original implementation is `FeedbackQueue`.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -82,11 +82,13 @@ public ListStateWithCache(
runtimeContext.getUserCodeClassLoader());
if (fraction > 0) {
MemoryManager memoryManager = containingTask.getEnvironment().getMemoryManager();
+ // TODO Adds a configuration to set the fraction of the memory that the
+ // ListStateWithCache can use.
segmentPool =
new LazyMemorySegmentPool(
containingTask,
memoryManager,
- memoryManager.computeNumberOfPages(fraction));
+ memoryManager.computeNumberOfPages(fraction) / 2);
Review Comment:
Just wondering, as `SpillableQueue` is only used in `HeadOperator` and `TailOperator`, how does it influence the `fraction` in `ListStateWithCache`?
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "Fanoid (via GitHub)" <gi...@apache.org>.
Fanoid commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1282923826
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -82,11 +82,13 @@ public ListStateWithCache(
runtimeContext.getUserCodeClassLoader());
if (fraction > 0) {
MemoryManager memoryManager = containingTask.getEnvironment().getMemoryManager();
+ // TODO Adds a configuration to set the fraction of the memory that the
+ // ListStateWithCache can use.
segmentPool =
new LazyMemorySegmentPool(
containingTask,
memoryManager,
- memoryManager.computeNumberOfPages(fraction));
+ memoryManager.computeNumberOfPages(fraction) / 2);
Review Comment:
The changes are acceptable to me. Thank you for your efforts in improving them!
--
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] lindong28 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1295838409
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/typeinfo/IterationRecordSerializer.java:
##########
@@ -106,7 +106,7 @@ public int getLength() {
@Override
public void serialize(IterationRecord<T> record, DataOutputView target) throws IOException {
target.writeByte((byte) record.getType().ordinal());
- serializerNumber(record.getEpoch(), target);
+ target.writeInt(record.getEpoch());
Review Comment:
It seems that this change will increase the average epoch overhead for common cases.
Can you explain why this change is needed?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
Review Comment:
It seems simpler to get pageSize via `memoryManager.getPageSize()`.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
+ }
+
+ void close() throws IOException {
+ output.clear();
+ List<MemorySegment> toRelease = target.close();
+ toRelease.addAll(segments);
+ memoryManager.release(toRelease);
+ segmentPool.clear();
+ }
+
+ private void resetSpillingBuffer() throws MemoryAllocationException {
+ this.segments = memoryManager.allocatePages(this, numPages);
+ this.segmentPool = new ListMemorySegmentPool(segments);
Review Comment:
Would it be simpler to use `ListMemorySegmentSource` (defined in flink-runtime) rather than ListMemorySegmentPool (defined in flink-table-runtime)?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackChannel.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.statefun.flink.core.feedback.FeedbackConsumer;
+import org.apache.flink.statefun.flink.core.feedback.SubtaskFeedbackKey;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Objects;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Single producer, single consumer channel, which can spill the records to disk when the in-memory
+ * buffer is full.
+ */
+@Internal
+public final class SpillableFeedbackChannel<T> implements Closeable {
+
+ /** The key that used to identify this channel. */
+ private final SubtaskFeedbackKey<T> key;
+
+ /** A single registered consumer. */
+ private final AtomicReference<ConsumerTask<T>> consumerRef = new AtomicReference<>();
+
+ /** The underlying queue used to hold the feedback results. */
+ private MpscQueue<T> queue;
+
+ /** Whether the feedback channel is initialized. */
+ private boolean initialized;
Review Comment:
Instead of adding this variable, would it be simpler to check whether `queue == null`?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
+ }
+
+ void close() throws IOException {
+ output.clear();
+ List<MemorySegment> toRelease = target.close();
+ toRelease.addAll(segments);
+ memoryManager.release(toRelease);
Review Comment:
Would it be simpler to just do `memoryManager.release(segments)`?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
Review Comment:
Would it be more readable to name it `buffer`?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
Review Comment:
Instead of repeatedly release and re-allocate this list of buffers, can we re-use the allocated buffers before/after the reset, similar to what is done in `SpillingResettableIterator#reset`?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
Review Comment:
close() is generally used to release all resources used by this instance and there should be no further invocation of any method of this instance after close() is invoked.
It would be useful to follow this pattern.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/MpscQueue.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.statefun.flink.core.queue.Lock;
+import org.apache.flink.statefun.flink.core.queue.Locks;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Multi producers single consumer fifo queue.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class MpscQueue<T> implements Closeable {
+ private final Lock lock = Locks.spinLock();
+
+ private SpillableFeedbackQueue<T> activeQueue;
+ private SpillableFeedbackQueue<T> standByQueue;
+
+ MpscQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.activeQueue =
+ new SpillableFeedbackQueue<>(
+ ioManager, memoryManager, serializer, inMemoryBufferSize / 2, pageSize);
+ this.standByQueue =
+ new SpillableFeedbackQueue<>(
+ ioManager, memoryManager, serializer, inMemoryBufferSize / 2, pageSize);
+ }
+
+ /**
+ * Adds an element to this (unbound) queue.
+ *
+ * @param element the element to add.
+ * @return the number of elements in the queue after the addition.
+ */
+ long add(T element) {
+ Preconditions.checkState(element instanceof StreamRecord);
+
+ lock.lockUninterruptibly();
Review Comment:
It would be useful to still use `final Lock lock = this.lock` as was done in flink-statefun's MpscQueue.
See https://stackoverflow.com/questions/13155860/why-does-jdk-sourcecode-take-a-final-copy-of-volatile-instances for an explanation.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/HeadOperator.java:
##########
@@ -422,16 +430,44 @@ public void close() throws Exception {
}
}
- private void registerFeedbackConsumer(Executor mailboxExecutor) {
+ private void registerFeedbackConsumer(Executor mailboxExecutor)
+ throws MemoryAllocationException {
+ StreamTask<?, ?> task = getContainingTask();
+
int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
int attemptNum = getRuntimeContext().getAttemptNumber();
FeedbackKey<StreamRecord<IterationRecord<?>>> feedbackKey =
OperatorUtils.createFeedbackKey(iterationId, feedbackIndex);
SubtaskFeedbackKey<StreamRecord<IterationRecord<?>>> key =
feedbackKey.withSubTaskIndex(indexOfThisSubtask, attemptNum);
- FeedbackChannelBroker broker = FeedbackChannelBroker.get();
- FeedbackChannel<StreamRecord<IterationRecord<?>>> channel = broker.getChannel(key);
- OperatorUtils.registerFeedbackConsumer(channel, this, mailboxExecutor);
+ SpillableFeedbackChannelBroker broker = SpillableFeedbackChannelBroker.get();
+ this.feedbackChannel = broker.getChannel(key);
Review Comment:
Would it be more readable to pass the initialization logic as a parameter of type `Consumer<SpillableFeedbackChannel>` to `getChannel(...)`?
We can do the same for TailOperator. The consumer will be invoked iff a new channel needs to be created.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "zhipeng93 (via GitHub)" <gi...@apache.org>.
zhipeng93 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1293245279
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/MpscQueue.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.statefun.flink.core.queue.Lock;
+import org.apache.flink.statefun.flink.core.queue.Locks;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Multi producers single consumer fifo queue.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+public final class MpscQueue<T> implements Closeable {
+ private final Lock lock = Locks.spinLock();
+
+ private SpillableFeedbackQueue<T> activeQueue;
+ private SpillableFeedbackQueue<T> standByQueue;
+
+ private volatile boolean isClosed;
+
+ public MpscQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.activeQueue =
+ new SpillableFeedbackQueue<>(
+ ioManager, memoryManager, serializer, inMemoryBufferSize / 2, pageSize);
+ this.standByQueue =
+ new SpillableFeedbackQueue<>(
+ ioManager, memoryManager, serializer, inMemoryBufferSize / 2, pageSize);
+ }
+
+ /**
+ * Adds an element to this (unbound) queue.
+ *
+ * @param element the element to add.
+ * @return the number of elements in the queue after the addition.
+ */
+ public long add(T element) {
+ Preconditions.checkState(element instanceof StreamRecord);
+
+ lock.lockUninterruptibly();
+ try {
+ SpillableFeedbackQueue<T> active = this.activeQueue;
+
+ active.add(element);
+ return active.size();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Atomically drains the queue.
+ *
+ * @return a batch of elements that obtained atomically from that queue.
+ */
+ public MutableObjectIterator<T> drainAll() {
+ lock.lockUninterruptibly();
+ try {
+ final SpillableFeedbackQueue<T> ready = this.activeQueue;
+ if (ready.size() == 0) {
+ return EmptyMutableObjectIterator.get();
+ }
+ this.activeQueue = this.standByQueue;
+ this.standByQueue = ready;
+ return ready.iterate();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void resetStandBy() throws Exception {
Review Comment:
It seems confusing for users to call `resetStandBy`. How about we rename this method as `resetAfterDrainAll()`?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackChannel.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.statefun.flink.core.feedback.FeedbackConsumer;
+import org.apache.flink.statefun.flink.core.feedback.SubtaskFeedbackKey;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Objects;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Single producer, single consumer channel, which can spill the records to disk when the in-memory
+ * buffer is full.
+ */
+@Internal
+public final class SpillableFeedbackChannel<T> implements Closeable {
+
+ /** The key that used to identify this channel. */
+ private final SubtaskFeedbackKey<T> key;
+
+ /** A single registered consumer. */
+ private final AtomicReference<ConsumerTask<T>> consumerRef = new AtomicReference<>();
+
+ /** The underlying queue used to hold the feedback results. */
+ private MpscQueue<T> queue;
+
+ /** Whether the feedback channel is initialized. */
+ private boolean initialized;
+
+ SpillableFeedbackChannel(SubtaskFeedbackKey<T> key) {
+ this.key = Objects.requireNonNull(key);
+ }
+
+ public void initialize(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.queue =
+ new MpscQueue<>(ioManager, memoryManager, serializer, inMemoryBufferSize, pageSize);
+ this.initialized = true;
Review Comment:
Is there a case that this object is not initialized?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/MpscQueue.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.statefun.flink.core.queue.Lock;
+import org.apache.flink.statefun.flink.core.queue.Locks;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Multi producers single consumer fifo queue.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+public final class MpscQueue<T> implements Closeable {
+ private final Lock lock = Locks.spinLock();
+
+ private SpillableFeedbackQueue<T> activeQueue;
+ private SpillableFeedbackQueue<T> standByQueue;
+
+ private volatile boolean isClosed;
+
+ public MpscQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.activeQueue =
+ new SpillableFeedbackQueue<>(
+ ioManager, memoryManager, serializer, inMemoryBufferSize / 2, pageSize);
+ this.standByQueue =
+ new SpillableFeedbackQueue<>(
+ ioManager, memoryManager, serializer, inMemoryBufferSize / 2, pageSize);
+ }
+
+ /**
+ * Adds an element to this (unbound) queue.
+ *
+ * @param element the element to add.
+ * @return the number of elements in the queue after the addition.
+ */
+ public long add(T element) {
+ Preconditions.checkState(element instanceof StreamRecord);
+
+ lock.lockUninterruptibly();
+ try {
+ SpillableFeedbackQueue<T> active = this.activeQueue;
+
+ active.add(element);
+ return active.size();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Atomically drains the queue.
+ *
+ * @return a batch of elements that obtained atomically from that queue.
+ */
+ public MutableObjectIterator<T> drainAll() {
+ lock.lockUninterruptibly();
+ try {
+ final SpillableFeedbackQueue<T> ready = this.activeQueue;
+ if (ready.size() == 0) {
+ return EmptyMutableObjectIterator.get();
+ }
+ this.activeQueue = this.standByQueue;
+ this.standByQueue = ready;
+ return ready.iterate();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void resetStandBy() throws Exception {
Review Comment:
nit: We may also make some of the methods in this PR package private.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/HeadOperator.java:
##########
@@ -498,6 +534,11 @@ public HeadOperatorStatus getStatus() {
return status;
}
+ @VisibleForTesting
+ public SpillableFeedbackChannel getFeedbackChannel() {
Review Comment:
Can we remove this method and close the feedback channel in `HeadOperator#close()`?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/MpscQueue.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.statefun.flink.core.queue.Lock;
+import org.apache.flink.statefun.flink.core.queue.Locks;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Multi producers single consumer fifo queue.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+public final class MpscQueue<T> implements Closeable {
+ private final Lock lock = Locks.spinLock();
+
+ private SpillableFeedbackQueue<T> activeQueue;
+ private SpillableFeedbackQueue<T> standByQueue;
+
+ private volatile boolean isClosed;
Review Comment:
Can you explain why do we need a `volatile isClosed` here?
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on PR #248:
URL: https://github.com/apache/flink-ml/pull/248#issuecomment-1679955970
@zhipeng93 Thanks for the review. 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] jiangxin369 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1298087863
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/typeinfo/IterationRecordSerializer.java:
##########
@@ -106,7 +106,7 @@ public int getLength() {
@Override
public void serialize(IterationRecord<T> record, DataOutputView target) throws IOException {
target.writeByte((byte) record.getType().ordinal());
- serializerNumber(record.getEpoch(), target);
+ target.writeInt(record.getEpoch());
Review Comment:
Currently, the `Integer.MAX_VALUE + 1` is passed because the termination check is in the `HeadOperator` . When the `TailOperator` processed the last epoch_watermark, it needs to increment the epoch which is already `Integer.MAX_VALUE`, and send the increased epoch to the `HeadOperator`. Then the `TerminatingHeadOperatorRecordProcessor#processFeedbackElement` would check if the epoch is equal to `Integer.MAX_VALUE + 1` to determine whether to terminate the iteration.
Of course, we can avoid it. Maybe simply change the last epoch_watermark number to `Integer.MAX_VALUE - 1`, but I don't think we need to make the code less readable 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: 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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1322717816
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
Review Comment:
After reading the code in Flink, I found that the `SerializedUpdateBuffer` is just what we need, so I replaced the SpillingBuffer with SerializedUpdateBuffer to avoid allocating memory multiple times.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1295375489
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/MpscQueue.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.statefun.flink.core.queue.Lock;
+import org.apache.flink.statefun.flink.core.queue.Locks;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Multi producers single consumer fifo queue.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+public final class MpscQueue<T> implements Closeable {
+ private final Lock lock = Locks.spinLock();
+
+ private SpillableFeedbackQueue<T> activeQueue;
+ private SpillableFeedbackQueue<T> standByQueue;
+
+ private volatile boolean isClosed;
Review Comment:
There are some tests like `HeadOperatorTest#testSnapshotAndRestoreBeforeRoundZeroFinish` that are closing the operator without processing any events. So the `ConsumerTask` may be scheduled after the operator is closed and still allocates managed memory, which would cause errors.
But in an actual Flink program, the above case would not happen. So yes, the `volatile isClosed` can be removed, and just need to make the current tests process all events before closing.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1294652323
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/HeadOperator.java:
##########
@@ -498,6 +534,11 @@ public HeadOperatorStatus getStatus() {
return status;
}
+ @VisibleForTesting
+ public SpillableFeedbackChannel getFeedbackChannel() {
Review Comment:
The feedback channel is used by both `HeadOperator` and `TailOperator`, so we cannot close it in `HeadOperator` in case of `TailOperator` is still using it.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/MpscQueue.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.util.EmptyMutableObjectIterator;
+import org.apache.flink.statefun.flink.core.queue.Lock;
+import org.apache.flink.statefun.flink.core.queue.Locks;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Multi producers single consumer fifo queue.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+public final class MpscQueue<T> implements Closeable {
+ private final Lock lock = Locks.spinLock();
+
+ private SpillableFeedbackQueue<T> activeQueue;
+ private SpillableFeedbackQueue<T> standByQueue;
+
+ private volatile boolean isClosed;
+
+ public MpscQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.activeQueue =
+ new SpillableFeedbackQueue<>(
+ ioManager, memoryManager, serializer, inMemoryBufferSize / 2, pageSize);
+ this.standByQueue =
+ new SpillableFeedbackQueue<>(
+ ioManager, memoryManager, serializer, inMemoryBufferSize / 2, pageSize);
+ }
+
+ /**
+ * Adds an element to this (unbound) queue.
+ *
+ * @param element the element to add.
+ * @return the number of elements in the queue after the addition.
+ */
+ public long add(T element) {
+ Preconditions.checkState(element instanceof StreamRecord);
+
+ lock.lockUninterruptibly();
+ try {
+ SpillableFeedbackQueue<T> active = this.activeQueue;
+
+ active.add(element);
+ return active.size();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Atomically drains the queue.
+ *
+ * @return a batch of elements that obtained atomically from that queue.
+ */
+ public MutableObjectIterator<T> drainAll() {
+ lock.lockUninterruptibly();
+ try {
+ final SpillableFeedbackQueue<T> ready = this.activeQueue;
+ if (ready.size() == 0) {
+ return EmptyMutableObjectIterator.get();
+ }
+ this.activeQueue = this.standByQueue;
+ this.standByQueue = ready;
+ return ready.iterate();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void resetStandBy() throws Exception {
Review Comment:
I agree. Now that I have changed these methods to package private, maybe it is not necessary to rename the `resetStandBy` because it is not visible to users anymore.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackChannel.java:
##########
@@ -0,0 +1,147 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.statefun.flink.core.feedback.FeedbackConsumer;
+import org.apache.flink.statefun.flink.core.feedback.SubtaskFeedbackKey;
+import org.apache.flink.util.IOUtils;
+import org.apache.flink.util.MutableObjectIterator;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Closeable;
+import java.util.Objects;
+import java.util.concurrent.Executor;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Single producer, single consumer channel, which can spill the records to disk when the in-memory
+ * buffer is full.
+ */
+@Internal
+public final class SpillableFeedbackChannel<T> implements Closeable {
+
+ /** The key that used to identify this channel. */
+ private final SubtaskFeedbackKey<T> key;
+
+ /** A single registered consumer. */
+ private final AtomicReference<ConsumerTask<T>> consumerRef = new AtomicReference<>();
+
+ /** The underlying queue used to hold the feedback results. */
+ private MpscQueue<T> queue;
+
+ /** Whether the feedback channel is initialized. */
+ private boolean initialized;
+
+ SpillableFeedbackChannel(SubtaskFeedbackKey<T> key) {
+ this.key = Objects.requireNonNull(key);
+ }
+
+ public void initialize(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.queue =
+ new MpscQueue<>(ioManager, memoryManager, serializer, inMemoryBufferSize, pageSize);
+ this.initialized = true;
Review Comment:
In the current code base, no. But this is a public utility class, I would like the developers who use this class not to forget to call the `initialize` method.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1322717816
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
Review Comment:
After reading the code in Flink, I found that the `SerializedUpdateBuffer` is just what we need, so I replaced the SpillingBuffer with SerializedUpdateBuffer which makes few changes and also avoids allocating memory multiple times.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1281676904
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/datacache/nonkeyed/ListStateWithCache.java:
##########
@@ -82,11 +82,13 @@ public ListStateWithCache(
runtimeContext.getUserCodeClassLoader());
if (fraction > 0) {
MemoryManager memoryManager = containingTask.getEnvironment().getMemoryManager();
+ // TODO Adds a configuration to set the fraction of the memory that the
+ // ListStateWithCache can use.
segmentPool =
new LazyMemorySegmentPool(
containingTask,
memoryManager,
- memoryManager.computeNumberOfPages(fraction));
+ memoryManager.computeNumberOfPages(fraction) / 2);
Review Comment:
That's a good point. I also think that the `iteration.memory.feedback-buffer.fraction` we just added is inappropriate because it is not defined in `ManagedMemoryUseCase` of Flink. So it would grab the managed memory with `ListStateWithCache` which may cause memory allocation failed.
I changed the implementation to also use the `ManagedMemoryUseCase.OPERATOR` to share the managed memory with `ListStateWithCache`. Would 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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "zhipeng93 (via GitHub)" <gi...@apache.org>.
zhipeng93 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1282877648
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/utils/DataStreamUtils.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.TableException;
+
+import java.util.Optional;
+
+/** Provides utility functions for {@link DataStream}. */
+public class DataStreamUtils {
+
+ /**
+ * Sets {Transformation#declareManagedMemoryUseCaseAtOperatorScope(ManagedMemoryUseCase, int)}
+ * using the given bytes for {@link ManagedMemoryUseCase#OPERATOR}.
+ *
+ * <p>This method is in reference to Flink's ExecNodeUtil.setManagedMemoryWeight. The provided
+ * bytes should be in the same scale as existing usage in Flink, for example,
+ * StreamExecWindowAggregate.WINDOW_AGG_MEMORY_RATIO.
+ */
+ public static <T> void setManagedMemoryWeight(DataStream<T> dataStream, long memoryBytes) {
+ if (memoryBytes > 0) {
+ final int weightInMebibyte = Math.max(1, (int) (memoryBytes >> 20));
+ final Optional<Integer> previousWeight =
+ dataStream
+ .getTransformation()
+ .declareManagedMemoryUseCaseAtOperatorScope(
+ ManagedMemoryUseCase.OPERATOR, weightInMebibyte);
+ if (previousWeight.isPresent()) {
Review Comment:
Is there a case that one operator would set managed memory twice?
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/utils/DataStreamUtils.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.table.api.TableException;
+
+import java.util.Optional;
+
+/** Provides utility functions for {@link DataStream}. */
+public class DataStreamUtils {
+
+ /**
+ * Sets {Transformation#declareManagedMemoryUseCaseAtOperatorScope(ManagedMemoryUseCase, int)}
+ * using the given bytes for {@link ManagedMemoryUseCase#OPERATOR}.
+ *
+ * <p>This method is in reference to Flink's ExecNodeUtil.setManagedMemoryWeight. The provided
+ * bytes should be in the same scale as existing usage in Flink, for example,
+ * StreamExecWindowAggregate.WINDOW_AGG_MEMORY_RATIO.
+ */
+ public static <T> void setManagedMemoryWeight(DataStream<T> dataStream, long memoryBytes) {
+ if (memoryBytes > 0) {
+ final int weightInMebibyte = Math.max(1, (int) (memoryBytes >> 20));
+ final Optional<Integer> previousWeight =
+ dataStream
+ .getTransformation()
+ .declareManagedMemoryUseCaseAtOperatorScope(
+ ManagedMemoryUseCase.OPERATOR, weightInMebibyte);
+ if (previousWeight.isPresent()) {
Review Comment:
Is there a case that one operator would set managed memory twice?
--
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] lindong28 commented on pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on PR #248:
URL: https://github.com/apache/flink-ml/pull/248#issuecomment-1717534003
Thanks for the update! 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] lindong28 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1297874568
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/typeinfo/IterationRecordSerializer.java:
##########
@@ -106,7 +106,7 @@ public int getLength() {
@Override
public void serialize(IterationRecord<T> record, DataOutputView target) throws IOException {
target.writeByte((byte) record.getType().ordinal());
- serializerNumber(record.getEpoch(), target);
+ target.writeInt(record.getEpoch());
Review Comment:
Regarding "the original serializerNumber cannot serialize Integer.MAX_VALUE + 1", can you explain why it is necessary to pass `Integer.MAX_VALUE + 1` as the serializerNumber? Is it possible to change the caller side code to avoid that?
I agree the 1-byte overhead is likely not a big deal for common cases where iteration is needed. On the other hand, if it is possible to change the caller side code as appropriate without any downside, then that approach is better than increasing the epoch overhead from 1 byte to 4 bytes.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/typeinfo/IterationRecordSerializer.java:
##########
@@ -106,7 +106,7 @@ public int getLength() {
@Override
public void serialize(IterationRecord<T> record, DataOutputView target) throws IOException {
target.writeByte((byte) record.getType().ordinal());
- serializerNumber(record.getEpoch(), target);
+ target.writeInt(record.getEpoch());
Review Comment:
Regarding "the original serializerNumber cannot serialize Integer.MAX_VALUE + 1", can you explain why it is necessary to pass `Integer.MAX_VALUE + 1` as the serializerNumber? Is it possible to change the caller side code to avoid that?
I agree the 1-byte overhead is likely not a big deal for common cases where iteration is needed. On the other hand, if it is possible to change the caller side code as appropriate without any downside, then that approach would be strictly better than increasing the epoch overhead from 1 byte to 4 bytes.
--
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] lindong28 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "lindong28 (via GitHub)" <gi...@apache.org>.
lindong28 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1297881781
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/HeadOperator.java:
##########
@@ -422,16 +430,44 @@ public void close() throws Exception {
}
}
- private void registerFeedbackConsumer(Executor mailboxExecutor) {
+ private void registerFeedbackConsumer(Executor mailboxExecutor)
+ throws MemoryAllocationException {
+ StreamTask<?, ?> task = getContainingTask();
+
int indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
int attemptNum = getRuntimeContext().getAttemptNumber();
FeedbackKey<StreamRecord<IterationRecord<?>>> feedbackKey =
OperatorUtils.createFeedbackKey(iterationId, feedbackIndex);
SubtaskFeedbackKey<StreamRecord<IterationRecord<?>>> key =
feedbackKey.withSubTaskIndex(indexOfThisSubtask, attemptNum);
- FeedbackChannelBroker broker = FeedbackChannelBroker.get();
- FeedbackChannel<StreamRecord<IterationRecord<?>>> channel = broker.getChannel(key);
- OperatorUtils.registerFeedbackConsumer(channel, this, mailboxExecutor);
+ SpillableFeedbackChannelBroker broker = SpillableFeedbackChannelBroker.get();
+ this.feedbackChannel = broker.getChannel(key);
Review Comment:
With the current implementation, both HeadOperator and TailOperator will get channel via `broker.getChannel(key)`. HeadOperator is responsible for initializing this channel by calling its `initialize()` API. TailOperator will encounter exception if the channel has not been initialized.
This approach effectively relies on the callers (i.e. HeadOperator and TailOperator) to coordinator how they should access this shared object. It would be simpler if the caller code does not rely on this coordination to work.
And this code is less readable. The reason is that `SpillableFeedbackChannel#getChannel` effectively means "getOrCreateChannel" since it creates a channel if it does not already exist for the given key. However, this channel can only be used after it is initialized, which is not guaranteed by `getChannel`. By passing the initialization logic as an extra parameter to `getChannel(..)`, this method will always return a usable channel. This can make the code more readable and more self-contained. And it can also address the issue mentioned above.
--
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 #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "zhipeng93 (via GitHub)" <gi...@apache.org>.
zhipeng93 commented on PR #248:
URL: https://github.com/apache/flink-ml/pull/248#issuecomment-1680296164
Thanks for the update. LGTM.
@lindong28 Do you want to take a 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] jiangxin369 commented on a diff in pull request #248: [FLINK-32704] Supports spilling to disk when feedback channel memory buffer is full
Posted by "jiangxin369 (via GitHub)" <gi...@apache.org>.
jiangxin369 commented on code in PR #248:
URL: https://github.com/apache/flink-ml/pull/248#discussion_r1297235534
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/typeinfo/IterationRecordSerializer.java:
##########
@@ -106,7 +106,7 @@ public int getLength() {
@Override
public void serialize(IterationRecord<T> record, DataOutputView target) throws IOException {
target.writeByte((byte) record.getType().ordinal());
- serializerNumber(record.getEpoch(), target);
+ target.writeInt(record.getEpoch());
Review Comment:
The original `serializerNumber` cannot serialize `Integer.MAX_VALUE + 1`, which is the epoch of the feedback record in the last iteration. In the previous implementation, the feedback records are just passed in memory and don't need to serialize/deserialize, so the bug doesn't appear.
I also did a benchmark to serialize 1 billion integers with `serializerNumber` and `DataOutputSerializer#writeInt`, their performances are quite close, so I think we can remove the current `serializerNumber` implementation.
##########
flink-ml-iteration/flink-ml-iteration-common/src/main/java/org/apache/flink/iteration/operator/feedback/SpillableFeedbackQueue.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.operator.feedback;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.runtime.io.disk.InputViewIterator;
+import org.apache.flink.runtime.io.disk.SpillingBuffer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.table.runtime.operators.sort.ListMemorySegmentPool;
+import org.apache.flink.util.MutableObjectIterator;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * * A queue that can spill the items to disks automatically when the memory buffer is full.
+ *
+ * @param <T> The element type.
+ */
+@Internal
+final class SpillableFeedbackQueue<T> {
+ private final DataOutputSerializer output = new DataOutputSerializer(256);
+ private final TypeSerializer<T> serializer;
+ private final IOManager ioManager;
+ private final MemoryManager memoryManager;
+ private final int numPages;
+
+ private List<MemorySegment> segments;
+ private ListMemorySegmentPool segmentPool;
+
+ private SpillingBuffer target;
+ private long size = 0L;
+
+ SpillableFeedbackQueue(
+ IOManager ioManager,
+ MemoryManager memoryManager,
+ TypeSerializer<T> serializer,
+ long inMemoryBufferSize,
+ long pageSize)
+ throws MemoryAllocationException {
+ this.serializer = Objects.requireNonNull(serializer);
+ this.ioManager = Objects.requireNonNull(ioManager);
+ this.memoryManager = Objects.requireNonNull(memoryManager);
+
+ this.numPages = (int) (inMemoryBufferSize / pageSize);
+ resetSpillingBuffer();
+ }
+
+ void add(T item) {
+ try {
+ output.clear();
+ serializer.serialize(item, output);
+ target.write(output.getSharedBuffer(), 0, output.length());
+ size++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ MutableObjectIterator<T> iterate() {
+ try {
+ DataInputView input = target.flip();
+ return new InputViewIterator<>(input, this.serializer);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ long size() {
+ return size;
+ }
+
+ void reset() throws Exception {
+ size = 0;
+ close();
+ resetSpillingBuffer();
+ }
+
+ void close() throws IOException {
+ output.clear();
+ List<MemorySegment> toRelease = target.close();
+ toRelease.addAll(segments);
+ memoryManager.release(toRelease);
Review Comment:
No, because every segment would be removed one by one from the `segments` when it is used, it can be seen in `ListMemorySegmentSource#nextSegment`. So the return value of `SpillingBuffer.close()` and the remaining `segments` together are the memories that should be freed.
--
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