You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/02/17 10:07:03 UTC
[GitHub] [flink] azagrebin opened a new pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
azagrebin opened a new pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109
## What is the purpose of the change
After #9747, managed memory is allocated from UNSAFE, not as direct nio buffers as before 1.10.
The releasing of segments released also underlying unsafe memory which is dangerous in general
as there can be still references to java objects giving access to the released memory. If this reference
ever leaks, the illegal memory access can result in memory corruption of other code parts w/o even segmentation fault.
The solution can be similar to how Java handles direct memory limit:
- underlying byte buffers of segments are registered to phantom reference queue with a Java GC cleaner which releases the unsafe memory
- all allocations and reservations are managed by a memory limit and an atomic available memory
- if available memory is not enough while reserving, the phantom reference queue processing is triggered to run hopefully discovered by GC cleaners
- memory can be released directly or in a GC cleaner
The GC is also sped up by nulling out byte buffer reference in `HybridMemorySegment#free` which is inaccessible anyways after freeing. Otherwise also a lot of tests, which hold accidental references to memory segments, have to be fixed to not hold them. The `MemoryManager#verifyEmpty` checks that everything can be GC'ed at the end of the tests and after slot closing in production to detect memory leaks if any other references are held, e.g. from `HybridMemorySegment#wrap`.
## Brief change log
- Simply MemoryManager by removing KeyedBudgetManager as the managed memory is only on-heap now
- Remove MemoryManager#AllocationRequest
- Implement custom unsafe memory control, similar to what JVM does to control direct memory limit
- Use AtomicLong to control allocated memory and optimistic allocation retry (like in nio.Bits#tryReserveMemory) and trigger GC cleaners if memory is unavailable
- adjust and add unit tests
## Verifying this change
unit tests
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410251612
##########
File path: flink-core/src/main/java/org/apache/flink/util/JavaGcCleanerWrapper.java
##########
@@ -70,35 +70,46 @@ public static Runnable create(Object owner, Runnable cleanOperation) {
return CLEANER_FACTORY.create(owner, cleanOperation);
}
+ public static boolean waitForGcToRunReadyCleaners() throws InterruptedException {
+ return CLEANER_FACTORY.waitForGcToRunReadyCleaners();
+ }
+
+ private static Class<?> findClass(String className, String errorMessage) {
+ try {
+ return Class.forName(className);
+ } catch (ClassNotFoundException e) {
+ throw new FlinkRuntimeException(errorMessage, e);
+ }
+ }
+
@FunctionalInterface
private interface CleanerProvider {
CleanerFactory createCleanerFactory() throws ClassNotFoundException;
}
- @FunctionalInterface
private interface CleanerFactory {
Runnable create(Object owner, Runnable cleanOperation);
+
+ boolean waitForGcToRunReadyCleaners() throws InterruptedException;
Review comment:
I think this method does not belong to the `CleanerFactory`. If you take a look at the implementations of the `CleanerFactory` methods, then one sees that both interface methods are completely disjunct. I would suggest to introduce a separate type for it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149255925 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
-->
## CI report:
* c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149255925) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410218003
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
##########
@@ -83,41 +73,47 @@
private final Map<Object, Set<MemorySegment>> allocatedSegments;
/** Reserved memory per memory owner. */
- private final Map<Object, Map<MemoryType, Long>> reservedMemory;
+ private final Map<Object, Long> reservedMemory;
+
+ private final long pageSize;
+
+ private final long totalMemorySize;
+
+ private final long totalNumberOfPages;
- private final KeyedBudgetManager<MemoryType> budgetByType;
+ private final AtomicLong availableMemorySize;
private final SharedResources sharedResources;
/** Flag whether the close() has already been invoked. */
private volatile boolean isShutDown;
/**
- * Creates a memory manager with the given memory types, capacity and given page size.
+ * Creates a memory manager with the given capacity and given page size.
*
- * @param memorySizeByType The total size of the memory to be managed by this memory manager for each type (heap / off-heap).
+ * @param memorySize The total size of the off-heap memory to be managed by this memory manager.
* @param pageSize The size of the pages handed out by the memory manager.
*/
- public MemoryManager(Map<MemoryType, Long> memorySizeByType, int pageSize) {
- for (Entry<MemoryType, Long> sizeForType : memorySizeByType.entrySet()) {
- sanityCheck(sizeForType.getValue(), pageSize, sizeForType.getKey());
- }
+ public MemoryManager(long memorySize, int pageSize) {
+ sanityCheck(memorySize, pageSize);
+ this.pageSize = pageSize;
+ this.totalMemorySize = memorySize;
+ this.totalNumberOfPages = memorySize / pageSize;
this.allocatedSegments = new ConcurrentHashMap<>();
this.reservedMemory = new ConcurrentHashMap<>();
- this.budgetByType = new KeyedBudgetManager<>(memorySizeByType, pageSize);
+ this.availableMemorySize = new AtomicLong(totalMemorySize);
this.sharedResources = new SharedResources();
- verifyIntTotalNumberOfPages(memorySizeByType, budgetByType.maxTotalNumberOfPages());
+ verifyIntTotalNumberOfPages(totalMemorySize, totalNumberOfPages);
LOG.debug(
"Initialized MemoryManager with total memory size {} ({}), page size {}.",
- budgetByType.totalAvailableBudget(),
- memorySizeByType,
+ memorySize,
+ memorySize,
Review comment:
I think we can remove on `memorySize` 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410264001
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
##########
@@ -637,44 +622,7 @@ public long computeMemorySize(double fraction) {
"The fraction of memory to allocate must within (0, 1], was: %s", fraction);
//noinspection NumericCastThatLosesPrecision
Review comment:
Please don't add suppressions for IntelliJ's inspections. I think here we are mixing dev specific tooling with Flink code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410269608
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/UnsafeMemoryBudget.java
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.memory;
+
+import org.apache.flink.util.JavaGcCleanerWrapper;
+
+import javax.annotation.Nonnegative;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Tracker of memory allocation and release within a custom limit.
+ *
+ * <p>This memory management in Flink uses the same approach as Java uses to allocate direct memory
+ * and release it upon GC but memory here can be also released directly with {@link #releaseMemory(long)}.
+ * The memory reservation {@link #reserveMemory(long)} tries firstly to run all phantom cleaners, created with
+ * {@link org.apache.flink.core.memory.MemoryUtils#createMemoryGcCleaner(Object, long, Runnable)},
+ * for objects which are ready to be garbage collected. If memory is still not available, it triggers GC and
+ * continues to process any ready cleaners making {@link #MAX_SLEEPS} attempts before throwing {@link OutOfMemoryError}.
+ */
+class UnsafeMemoryBudget {
+ // max. number of sleeps during try-reserving with exponentially
+ // increasing delay before throwing OutOfMemoryError:
+ // 1, 2, 4, 8, 16, 32, 64, 128, 256, 512 (total 1023 ms ~ 1 s)
+ // which means that OOME will be thrown after 0.5 s of trying
+ private static final int MAX_SLEEPS = 10;
+ private static final int RETRIGGER_GC_AFTER_SLEEPS = 9; // ~ 0.5 sec
+
+ private final long totalMemorySize;
+
+ private final AtomicLong availableMemorySize;
+
+ UnsafeMemoryBudget(long totalMemorySize) {
+ this.totalMemorySize = totalMemorySize;
+ this.availableMemorySize = new AtomicLong(totalMemorySize);
+ }
+
+ long getTotalMemorySize() {
+ return totalMemorySize;
+ }
+
+ long getAvailableMemorySize() {
+ return availableMemorySize.get();
+ }
+
+ boolean verifyEmpty() {
+ try {
+ reserveMemory(totalMemorySize);
+ } catch (MemoryReservationException e) {
+ return false;
+ }
+ releaseMemory(totalMemorySize);
+ return availableMemorySize.get() == totalMemorySize;
+ }
+
+ /**
+ * Reserve memory of certain size if it is available.
+ *
+ * <p>Adjusted version of {@link java.nio.Bits#reserveMemory(long, int)} taken from Java 11.
+ */
+ @SuppressWarnings({"OverlyComplexMethod", "JavadocReference", "NestedTryStatement"})
+ void reserveMemory(long size) throws MemoryReservationException {
+ long availableOrReserved = tryReserveMemory(size);
+ // optimist!
+ if (availableOrReserved >= size) {
+ return;
+ }
+
+ boolean interrupted = false;
+ try {
+
+ // Retry allocation until success or there are no more
+ // references (including Cleaners that might free direct
+ // buffer memory) to process and allocation still fails.
+ boolean refprocActive;
+ do {
+ try {
+ refprocActive = JavaGcCleanerWrapper.waitForGcToRunReadyCleaners();
+ } catch (InterruptedException e) {
+ // Defer interrupts and keep trying.
+ interrupted = true;
+ refprocActive = true;
+ }
+ availableOrReserved = tryReserveMemory(size);
+ if (availableOrReserved >= size) {
+ return;
+ }
+ } while (refprocActive);
+
+ // trigger VM's Reference processing
+ //noinspection CallToSystemGC
+ System.gc();
+
+ // A retry loop with exponential back-off delays.
+ // Sometimes it would suffice to give up once reference
+ // processing is complete. But if there are many threads
+ // competing for memory, this gives more opportunities for
+ // any given thread to make progress. In particular, this
+ // seems to be enough for a stress test like
+ // DirectBufferAllocTest to (usually) succeed, while
+ // without it that test likely fails. Since failure here
+ // ends in OOME, there's no need to hurry.
Review comment:
Where are we throwing an OOM?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410268774
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/UnsafeMemoryBudget.java
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.memory;
+
+import org.apache.flink.util.JavaGcCleanerWrapper;
+
+import javax.annotation.Nonnegative;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Tracker of memory allocation and release within a custom limit.
+ *
+ * <p>This memory management in Flink uses the same approach as Java uses to allocate direct memory
+ * and release it upon GC but memory here can be also released directly with {@link #releaseMemory(long)}.
+ * The memory reservation {@link #reserveMemory(long)} tries firstly to run all phantom cleaners, created with
+ * {@link org.apache.flink.core.memory.MemoryUtils#createMemoryGcCleaner(Object, long, Runnable)},
+ * for objects which are ready to be garbage collected. If memory is still not available, it triggers GC and
+ * continues to process any ready cleaners making {@link #MAX_SLEEPS} attempts before throwing {@link OutOfMemoryError}.
+ */
+class UnsafeMemoryBudget {
+ // max. number of sleeps during try-reserving with exponentially
+ // increasing delay before throwing OutOfMemoryError:
+ // 1, 2, 4, 8, 16, 32, 64, 128, 256, 512 (total 1023 ms ~ 1 s)
+ // which means that OOME will be thrown after 0.5 s of trying
+ private static final int MAX_SLEEPS = 10;
+ private static final int RETRIGGER_GC_AFTER_SLEEPS = 9; // ~ 0.5 sec
+
+ private final long totalMemorySize;
+
+ private final AtomicLong availableMemorySize;
+
+ UnsafeMemoryBudget(long totalMemorySize) {
+ this.totalMemorySize = totalMemorySize;
+ this.availableMemorySize = new AtomicLong(totalMemorySize);
+ }
+
+ long getTotalMemorySize() {
+ return totalMemorySize;
+ }
+
+ long getAvailableMemorySize() {
+ return availableMemorySize.get();
+ }
+
+ boolean verifyEmpty() {
+ try {
+ reserveMemory(totalMemorySize);
+ } catch (MemoryReservationException e) {
+ return false;
+ }
+ releaseMemory(totalMemorySize);
+ return availableMemorySize.get() == totalMemorySize;
+ }
+
+ /**
+ * Reserve memory of certain size if it is available.
+ *
+ * <p>Adjusted version of {@link java.nio.Bits#reserveMemory(long, int)} taken from Java 11.
+ */
+ @SuppressWarnings({"OverlyComplexMethod", "JavadocReference", "NestedTryStatement"})
+ void reserveMemory(long size) throws MemoryReservationException {
+ long availableOrReserved = tryReserveMemory(size);
+ // optimist!
+ if (availableOrReserved >= size) {
+ return;
+ }
+
+ boolean interrupted = false;
+ try {
+
+ // Retry allocation until success or there are no more
+ // references (including Cleaners that might free direct
+ // buffer memory) to process and allocation still fails.
+ boolean refprocActive;
+ do {
+ try {
+ refprocActive = JavaGcCleanerWrapper.waitForGcToRunReadyCleaners();
+ } catch (InterruptedException e) {
+ // Defer interrupts and keep trying.
+ interrupted = true;
+ refprocActive = true;
+ }
+ availableOrReserved = tryReserveMemory(size);
+ if (availableOrReserved >= size) {
+ return;
+ }
+ } while (refprocActive);
+
+ // trigger VM's Reference processing
+ //noinspection CallToSystemGC
+ System.gc();
+
+ // A retry loop with exponential back-off delays.
+ // Sometimes it would suffice to give up once reference
+ // processing is complete. But if there are many threads
+ // competing for memory, this gives more opportunities for
+ // any given thread to make progress. In particular, this
+ // seems to be enough for a stress test like
+ // DirectBufferAllocTest to (usually) succeed, while
+ // without it that test likely fails. Since failure here
+ // ends in OOME, there's no need to hurry.
+ long sleepTime = 1;
+ int sleeps = 0;
+ while (true) {
+ availableOrReserved = tryReserveMemory(size);
+ if (availableOrReserved >= size) {
+ return;
+ }
+ if (sleeps >= MAX_SLEEPS) {
+ break;
+ }
+ if (sleeps >= RETRIGGER_GC_AFTER_SLEEPS) {
+ // trigger again VM's Reference processing if we have to wait longer
+ //noinspection CallToSystemGC
Review comment:
IntelliJ inspection
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410268864
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/UnsafeMemoryBudget.java
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.memory;
+
+import org.apache.flink.util.JavaGcCleanerWrapper;
+
+import javax.annotation.Nonnegative;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Tracker of memory allocation and release within a custom limit.
+ *
+ * <p>This memory management in Flink uses the same approach as Java uses to allocate direct memory
+ * and release it upon GC but memory here can be also released directly with {@link #releaseMemory(long)}.
+ * The memory reservation {@link #reserveMemory(long)} tries firstly to run all phantom cleaners, created with
+ * {@link org.apache.flink.core.memory.MemoryUtils#createMemoryGcCleaner(Object, long, Runnable)},
+ * for objects which are ready to be garbage collected. If memory is still not available, it triggers GC and
+ * continues to process any ready cleaners making {@link #MAX_SLEEPS} attempts before throwing {@link OutOfMemoryError}.
+ */
+class UnsafeMemoryBudget {
+ // max. number of sleeps during try-reserving with exponentially
+ // increasing delay before throwing OutOfMemoryError:
+ // 1, 2, 4, 8, 16, 32, 64, 128, 256, 512 (total 1023 ms ~ 1 s)
+ // which means that OOME will be thrown after 0.5 s of trying
+ private static final int MAX_SLEEPS = 10;
+ private static final int RETRIGGER_GC_AFTER_SLEEPS = 9; // ~ 0.5 sec
+
+ private final long totalMemorySize;
+
+ private final AtomicLong availableMemorySize;
+
+ UnsafeMemoryBudget(long totalMemorySize) {
+ this.totalMemorySize = totalMemorySize;
+ this.availableMemorySize = new AtomicLong(totalMemorySize);
+ }
+
+ long getTotalMemorySize() {
+ return totalMemorySize;
+ }
+
+ long getAvailableMemorySize() {
+ return availableMemorySize.get();
+ }
+
+ boolean verifyEmpty() {
+ try {
+ reserveMemory(totalMemorySize);
+ } catch (MemoryReservationException e) {
+ return false;
+ }
+ releaseMemory(totalMemorySize);
+ return availableMemorySize.get() == totalMemorySize;
+ }
+
+ /**
+ * Reserve memory of certain size if it is available.
+ *
+ * <p>Adjusted version of {@link java.nio.Bits#reserveMemory(long, int)} taken from Java 11.
+ */
+ @SuppressWarnings({"OverlyComplexMethod", "JavadocReference", "NestedTryStatement"})
+ void reserveMemory(long size) throws MemoryReservationException {
+ long availableOrReserved = tryReserveMemory(size);
+ // optimist!
+ if (availableOrReserved >= size) {
+ return;
+ }
+
+ boolean interrupted = false;
+ try {
+
+ // Retry allocation until success or there are no more
+ // references (including Cleaners that might free direct
+ // buffer memory) to process and allocation still fails.
+ boolean refprocActive;
+ do {
+ try {
+ refprocActive = JavaGcCleanerWrapper.waitForGcToRunReadyCleaners();
+ } catch (InterruptedException e) {
+ // Defer interrupts and keep trying.
+ interrupted = true;
+ refprocActive = true;
+ }
+ availableOrReserved = tryReserveMemory(size);
+ if (availableOrReserved >= size) {
+ return;
+ }
+ } while (refprocActive);
+
+ // trigger VM's Reference processing
+ //noinspection CallToSystemGC
+ System.gc();
+
+ // A retry loop with exponential back-off delays.
+ // Sometimes it would suffice to give up once reference
+ // processing is complete. But if there are many threads
+ // competing for memory, this gives more opportunities for
+ // any given thread to make progress. In particular, this
+ // seems to be enough for a stress test like
+ // DirectBufferAllocTest to (usually) succeed, while
+ // without it that test likely fails. Since failure here
+ // ends in OOME, there's no need to hurry.
+ long sleepTime = 1;
+ int sleeps = 0;
+ while (true) {
+ availableOrReserved = tryReserveMemory(size);
+ if (availableOrReserved >= size) {
+ return;
+ }
+ if (sleeps >= MAX_SLEEPS) {
+ break;
+ }
+ if (sleeps >= RETRIGGER_GC_AFTER_SLEEPS) {
+ // trigger again VM's Reference processing if we have to wait longer
+ //noinspection CallToSystemGC
+ System.gc();
+ }
+ try {
+ if (!JavaGcCleanerWrapper.waitForGcToRunReadyCleaners()) {
+ //noinspection BusyWait
Review comment:
IntelliJ inspection
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410255110
##########
File path: flink-core/src/main/java/org/apache/flink/util/JavaGcCleanerWrapper.java
##########
@@ -254,5 +339,31 @@ public Runnable create(Object owner, Runnable cleanupOperation) {
}
};
}
+
+ @Override
+ public boolean waitForGcToRunReadyCleaners() throws InterruptedException {
+ Object javaLangRefAccess = getJavaLangRefAccess();
+ try {
+ return (Boolean) waitForReferenceProcessingMethod.invoke(javaLangRefAccess);
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ if (e.getCause() instanceof InterruptedException) {
+ //noinspection ThrowInsideCatchBlockWhichIgnoresCaughtException
+ throw (InterruptedException) e.getCause();
+ }
+ String message = "FATAL UNEXPECTED - Failed to invoke JavaLangRefAccess#waitForReferenceProcessing";
+ LOG.error(message, e);
+ throw new Error(message, e);
+ }
+ }
+
+ private Object getJavaLangRefAccess() {
+ try {
+ return getJavaLangRefAccessMethod.invoke(null);
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ String message = "FATAL UNEXPECTED - Failed to invoke SharedSecrets#getJavaLangRefAccess";
+ LOG.error(message, e);
+ throw new Error(message, e);
+ }
+ }
Review comment:
It looks to me as if these methods are more or less the same as the ones used in `LegacyCleanerFactory`. I think we can introduce a single type which is instantiated with the different `Methods` and then used for Java 9 and Java < 9.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410262958
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
##########
@@ -387,12 +377,7 @@ public void releaseAll(Object owner) {
}
// free each segment
- long releasedMemory = 0L;
- for (MemorySegment segment : segments) {
- segment.free();
- releasedMemory += getPageSize();
- }
- releaseMemory(releasedMemory);
+ segments.forEach(MemorySegment::free);
Review comment:
I would suggest to follow the KISS principle: Keep it simple, stupid:
```suggestion
for (MemorySegment segment: segments) {
segment.free();
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
azagrebin commented on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-588878235
I commented more in [FLINK-15094](https://issues.apache.org/jira/browse/FLINK-15094) about warning problem. There may be other workarounds to silence the warning.
Also some more thoughts about the netty pooling approach.
It contributes to direct memory limit which makes it hard to use interchangeably the native memory for managed batch and stream RocksDB.
If we do not want to use cleaners (this PR, btw they also look like being an official tool in future Java), netty also uses reference counting. In this case to not rely on GC, every time memory segment issues a ByteBuffer view and increment the ref counter, the user code has to decrement it at some point and ensure that the ByteBuffer view will never be used afterwards. The ByteBuffer view can be subsequently duplicated and/or sliced without contributing to ref counting.
I can look more into this, maybe I overlook something.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
azagrebin commented on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-587408504
@StephanEwen
Not sure, these changes are directly related to the warnings. The `HybridMemorySegment` and the GC cleaner of its unsafe memory does not strictly need `java.nio.ByteBuffer`s. It could be any other object handle (just private `Object` or `HybridMemorySegment` itself) which reference is kept in java code.
We illegally use the private constructor mostly to convert the unsafe `HybridMemorySegment` to the `java.nio.ByteBuffer` API in `HybridMemorySegment#getOffHeapBuffer` and `HybridMemorySegment#wrap` for network and file IO. This is the root case. An alternative to the private constructor could be our internal implementation of `java.nio.ByteBuffer`, which could be similar to `DirectByteBuffer`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
azagrebin edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-587491324
> the NIO stack can only access the memory directly if it finds a DirectByteBuffer.
Ok then true, that workaround will not work
> Instead, they start with a big DirectByteBuffer (arena or chunk)
This might work. Just if I understand correctly, this will contribute to JVM direct memory limit then. I thought, in general we wanted to keep it only for external/user code. Depending on how much memory pre-allocated and then (if) (de)allocated more, it will interfere with the user code more or less.
I think `MemorySegments`, allocated with this pooling approach, can still benefit from the GC cleaner implemented in this PR, just the GC scheduled release operation will be a bit different.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241",
"triggerID" : "c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31",
"triggerType" : "PUSH"
}, {
"hash" : "c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149255925",
"triggerID" : "c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31",
"triggerType" : "PUSH"
}, {
"hash" : "c71e8213f389b56ba642137c278c35272af3d58c",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149430663",
"triggerID" : "c71e8213f389b56ba642137c278c35272af3d58c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149255925) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241)
* c71e8213f389b56ba642137c278c35272af3d58c Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149430663)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410252956
##########
File path: flink-core/src/main/java/org/apache/flink/util/JavaGcCleanerWrapper.java
##########
@@ -218,6 +278,22 @@ private static Method getCleanMethod(Class<?> cleanableClass) {
}
}
+ private static Method getSharedSecretsGetJavaLangRefAccessMethod(Class<?> sharedSecretsClass) {
+ try {
+ return sharedSecretsClass.getMethod("getJavaLangRefAccess");
+ } catch (NoSuchMethodException e) {
+ throw new FlinkRuntimeException("Failed to find Java 9 SharedSecrets#getJavaLangRefAccess method", e);
+ }
+ }
+
+ private static Method getJavaLangRefAccessWaitForReferenceProcessing(Class<?> javaLangRefAccessClass) {
+ try {
+ return javaLangRefAccessClass.getMethod("waitForReferenceProcessing");
+ } catch (NoSuchMethodException e) {
+ throw new FlinkRuntimeException("Failed to find Java 9 JavaLangRefAccess#waitForReferenceProcessing method", e);
+ }
+ }
Review comment:
Same here with the deduplication. Maybe it's even across the different `CleanerProvider` implementations possible.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] StephanEwen commented on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-587373973
Could you share some thoughts in how this relates to future efforts to avoid "illegal reflective access" warnings on Java 9+?
It seems to be hard to have any interplay between "manually allocated native memory" and DirectByteBuffers in Java9+ without these warnings, and I assume that the solution to that is connected to the changes 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] StephanEwen commented on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-587415461
> An alternative to the private constructor could be our internal implementation of java.nio.ByteBuffer, which could be similar to DirectByteBuffer but this is a bit tedious. Nothing else comes into my mind atm.
I think this does not work. aside from the fact that this sounds like a very high-touch solution, the main reason why memory segment needs to interoperate with ByteBuffer is for I/O, and the NIO stack can only access the memory directly if it finds a DirectByteBuffer.
> Not sure, these changes are directly related to the warnings.
They may be related. If we want to keep `MemorySegment` and `DirectByteBuffer` interoperability (for the above reason), then we may have to consider that in the GC design. For example, Netty seems to not use unsafe memory allocation and manual cleaners any more in Java 9+. Instead, they start with a big DirectByteBuffer (arena or chunk) from which they slice smaller buffers and then turn them into ByteBufs. So the starting/entry point is always `ByteBuffer`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
azagrebin edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-587491324
> the NIO stack can only access the memory directly if it finds a DirectByteBuffer.
Ok then true, that workaround will not work
> Instead, they start with a big DirectByteBuffer (arena or chunk)
This may work. Just if I understand correctly, this will contribute to JVM direct memory limit then. I thought, in general we wanted to keep it only for external/user code. Depending on how much memory pre-allocated and then (if) (de)allocated more, it will interfere with the user code more or less.
I think `MemorySegments`, allocated with this pooling approach, can still benefit from the GC cleaner implemented in this PR, just the GC scheduled release operation will be a bit different.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410237975
##########
File path: flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java
##########
@@ -98,6 +104,8 @@
public ByteBuffer getOffHeapBuffer() {
if (offHeapBuffer != null) {
return offHeapBuffer;
+ } else if (address <= addressLimit) {
Review comment:
```suggestion
} else if (isFreed()) {
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:PENDING URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:PENDING URL:https://travis-ci.com/flink-ci/flink/builds/149255925 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
-->
## CI report:
* c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/149255925) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410243332
##########
File path: flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
##########
@@ -95,7 +96,12 @@ public static MemorySegment allocateUnpooledOffHeapMemory(int size) {
*/
public static MemorySegment allocateUnpooledOffHeapMemory(int size, Object owner) {
ByteBuffer memory = ByteBuffer.allocateDirect(size);
- return new HybridMemorySegment(memory, owner, null);
+ return new HybridMemorySegment(memory, owner);
+ }
+
+ @VisibleForTesting
+ public static MemorySegment allocateOffHeapUnsafeMemory(int size) {
+ return allocateOffHeapUnsafeMemory(size, null, () -> {});
Review comment:
Not sure but won't this call create a new anonymous class for the lambda for every call? What about creating one `static final Runnable NO_OP = () -> {};` which is used 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586914869
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 14446ad529ad989dfd0648b481cd405f0407296a (Mon Feb 17 10:09:17 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
azagrebin edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-588878235
I commented more in [FLINK-15094](https://issues.apache.org/jira/browse/FLINK-15094) about warning problem. There may be other workarounds to silence the warning.
Also some more thoughts about the netty pooling approach.
It contributes to direct memory limit which makes it hard to use interchangeably the native memory for managed batch and stream RocksDB.
If we do not want to use cleaners (this PR, btw they also look like being an official tool in future Java), netty also uses reference counting. In this case to not rely on GC, every time memory segment issues a ByteBuffer view and increments the ref counter, the user code has to explicitly decrement it at some point and ensure that the ByteBuffer view will never be used afterwards. It is hard because the ByteBuffer view can be subsequently duplicated and/or sliced without contributing to ref counting.
I can look more into this, maybe I overlook something.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149255925 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c71e8213f389b56ba642137c278c35272af3d58c Status:UNKNOWN URL:TBD TriggerType:PUSH TriggerID:c71e8213f389b56ba642137c278c35272af3d58c
-->
## CI report:
* c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149255925) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241)
* c71e8213f389b56ba642137c278c35272af3d58c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:PENDING URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149255925 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
-->
## CI report:
* c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149255925) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:UNKNOWN URL:TBD TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
-->
## CI report:
* c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410234004
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
##########
@@ -779,20 +735,12 @@ public int getNumberOfPages() {
return numberOfPages;
}
- public Set<MemoryType> getTypes() {
- return Collections.unmodifiableSet(types);
- }
-
public static Builder newBuilder(Object owner) {
return new Builder(owner);
}
- public static AllocationRequest ofAllTypes(Object owner, int numberOfPages) {
- return newBuilder(owner).ofAllTypes().numberOfPages(numberOfPages).build();
- }
-
- public static AllocationRequest ofType(Object owner, int numberOfPages, MemoryType type) {
- return newBuilder(owner).ofType(type).numberOfPages(numberOfPages).build();
+ public static AllocationRequest forOf(Object owner, int numberOfPages) {
Review comment:
I'm a bit unsure about the name of this method. Maybe rename it into `for` and don't use static imports in the test but write instead `AllocationRequest.for(owner, numberOfPages)`. I think this would be a bit clearer.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241",
"triggerID" : "c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31",
"triggerType" : "PUSH"
}, {
"hash" : "c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31",
"status" : "DELETED",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149255925",
"triggerID" : "c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31",
"triggerType" : "PUSH"
}, {
"hash" : "c71e8213f389b56ba642137c278c35272af3d58c",
"status" : "FAILURE",
"url" : "https://travis-ci.com/flink-ci/flink/builds/149430663",
"triggerID" : "c71e8213f389b56ba642137c278c35272af3d58c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c71e8213f389b56ba642137c278c35272af3d58c Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149430663)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
azagrebin edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-587408504
@StephanEwen
Not sure, these changes are directly related to the warnings. The `HybridMemorySegment` and the GC cleaner of its unsafe memory does not strictly need `java.nio.ByteBuffer`s. It could be any other object handle (just private `Object` or `HybridMemorySegment` itself) which reference is kept in java code.
We illegally use the private constructor mostly to convert the unsafe `HybridMemorySegment` to the `java.nio.ByteBuffer` API in `HybridMemorySegment#getOffHeapBuffer` and `HybridMemorySegment#wrap` for network and file IO. This is the root case. An alternative to the private constructor could be our internal implementation of `java.nio.ByteBuffer`, which could be similar to `DirectByteBuffer` but this is a bit tedious. Nothing else comes into my mind atm.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410285430
##########
File path: flink-core/src/main/java/org/apache/flink/util/JavaGcCleanerWrapper.java
##########
@@ -70,35 +70,46 @@ public static Runnable create(Object owner, Runnable cleanOperation) {
return CLEANER_FACTORY.create(owner, cleanOperation);
}
+ public static boolean waitForGcToRunReadyCleaners() throws InterruptedException {
+ return CLEANER_FACTORY.waitForGcToRunReadyCleaners();
+ }
+
+ private static Class<?> findClass(String className, String errorMessage) {
+ try {
+ return Class.forName(className);
+ } catch (ClassNotFoundException e) {
+ throw new FlinkRuntimeException(errorMessage, e);
+ }
+ }
+
@FunctionalInterface
private interface CleanerProvider {
CleanerFactory createCleanerFactory() throws ClassNotFoundException;
}
- @FunctionalInterface
private interface CleanerFactory {
Runnable create(Object owner, Runnable cleanOperation);
+
+ boolean waitForGcToRunReadyCleaners() throws InterruptedException;
Review comment:
Why not calling this method `tryRunPendingCleaners`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410244991
##########
File path: flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
##########
@@ -107,12 +113,14 @@ public static MemorySegment allocateUnpooledOffHeapMemory(int size, Object owner
*
* @param size The size of the off-heap unsafe memory segment to allocate.
* @param owner The owner to associate with the off-heap unsafe memory segment.
+ * @param customGcCleaner A custom action to run upon calling GC cleaner.
* @return A new memory segment, backed by off-heap unsafe memory.
*/
- public static MemorySegment allocateOffHeapUnsafeMemory(int size, Object owner) {
+ public static MemorySegment allocateOffHeapUnsafeMemory(int size, Object owner, Runnable customGcCleaner) {
Review comment:
```suggestion
public static MemorySegment allocateOffHeapUnsafeMemory(int size, Object owner, Runnable customCleanupAction) {
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149255925 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c71e8213f389b56ba642137c278c35272af3d58c Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149430663 TriggerType:PUSH TriggerID:c71e8213f389b56ba642137c278c35272af3d58c
-->
## CI report:
* c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149255925) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241)
* c71e8213f389b56ba642137c278c35272af3d58c Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149430663)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410254078
##########
File path: flink-core/src/main/java/org/apache/flink/util/JavaGcCleanerWrapper.java
##########
@@ -254,5 +339,31 @@ public Runnable create(Object owner, Runnable cleanupOperation) {
}
};
}
+
+ @Override
+ public boolean waitForGcToRunReadyCleaners() throws InterruptedException {
+ Object javaLangRefAccess = getJavaLangRefAccess();
+ try {
+ return (Boolean) waitForReferenceProcessingMethod.invoke(javaLangRefAccess);
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ if (e.getCause() instanceof InterruptedException) {
+ //noinspection ThrowInsideCatchBlockWhichIgnoresCaughtException
+ throw (InterruptedException) e.getCause();
+ }
Review comment:
Why is it required to handle the `InterruptedException` specifically?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410252210
##########
File path: flink-core/src/main/java/org/apache/flink/util/JavaGcCleanerWrapper.java
##########
@@ -117,6 +128,22 @@ private static Method getCleanerCleanMethod(Class<?> cleanerClass) {
}
}
+ private static Method getSharedSecretsGetJavaLangRefAccessMethod(Class<?> sharedSecretsClass) {
+ try {
+ return sharedSecretsClass.getMethod("getJavaLangRefAccess");
+ } catch (NoSuchMethodException e) {
+ throw new FlinkRuntimeException("Failed to find Java legacy SharedSecrets#getJavaLangRefAccess method", e);
+ }
+ }
+
+ private static Method getJavaLangRefAccessTryHandlePendingReference(Class<?> javaLangRefAccessClass) {
+ try {
+ return javaLangRefAccessClass.getMethod("tryHandlePendingReference");
+ } catch (NoSuchMethodException e) {
+ throw new FlinkRuntimeException("Failed to find Java legacy JavaLangRefAccess#tryHandlePendingReference method", e);
+ }
+ }
Review comment:
I think these two methods could be deduplicated via `findMethod(Class<?> class, String methodName)`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410267635
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/UnsafeMemoryBudget.java
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.memory;
+
+import org.apache.flink.util.JavaGcCleanerWrapper;
+
+import javax.annotation.Nonnegative;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Tracker of memory allocation and release within a custom limit.
+ *
+ * <p>This memory management in Flink uses the same approach as Java uses to allocate direct memory
+ * and release it upon GC but memory here can be also released directly with {@link #releaseMemory(long)}.
+ * The memory reservation {@link #reserveMemory(long)} tries firstly to run all phantom cleaners, created with
+ * {@link org.apache.flink.core.memory.MemoryUtils#createMemoryGcCleaner(Object, long, Runnable)},
+ * for objects which are ready to be garbage collected. If memory is still not available, it triggers GC and
+ * continues to process any ready cleaners making {@link #MAX_SLEEPS} attempts before throwing {@link OutOfMemoryError}.
+ */
+class UnsafeMemoryBudget {
+ // max. number of sleeps during try-reserving with exponentially
+ // increasing delay before throwing OutOfMemoryError:
+ // 1, 2, 4, 8, 16, 32, 64, 128, 256, 512 (total 1023 ms ~ 1 s)
+ // which means that OOME will be thrown after 0.5 s of trying
+ private static final int MAX_SLEEPS = 10;
+ private static final int RETRIGGER_GC_AFTER_SLEEPS = 9; // ~ 0.5 sec
+
+ private final long totalMemorySize;
+
+ private final AtomicLong availableMemorySize;
+
+ UnsafeMemoryBudget(long totalMemorySize) {
+ this.totalMemorySize = totalMemorySize;
+ this.availableMemorySize = new AtomicLong(totalMemorySize);
+ }
+
+ long getTotalMemorySize() {
+ return totalMemorySize;
+ }
+
+ long getAvailableMemorySize() {
+ return availableMemorySize.get();
+ }
+
+ boolean verifyEmpty() {
+ try {
+ reserveMemory(totalMemorySize);
+ } catch (MemoryReservationException e) {
+ return false;
+ }
+ releaseMemory(totalMemorySize);
+ return availableMemorySize.get() == totalMemorySize;
+ }
+
+ /**
+ * Reserve memory of certain size if it is available.
+ *
+ * <p>Adjusted version of {@link java.nio.Bits#reserveMemory(long, int)} taken from Java 11.
+ */
+ @SuppressWarnings({"OverlyComplexMethod", "JavadocReference", "NestedTryStatement"})
+ void reserveMemory(long size) throws MemoryReservationException {
+ long availableOrReserved = tryReserveMemory(size);
+ // optimist!
+ if (availableOrReserved >= size) {
+ return;
+ }
+
+ boolean interrupted = false;
+ try {
+
+ // Retry allocation until success or there are no more
+ // references (including Cleaners that might free direct
+ // buffer memory) to process and allocation still fails.
+ boolean refprocActive;
+ do {
+ try {
+ refprocActive = JavaGcCleanerWrapper.waitForGcToRunReadyCleaners();
+ } catch (InterruptedException e) {
+ // Defer interrupts and keep trying.
+ interrupted = true;
+ refprocActive = true;
+ }
+ availableOrReserved = tryReserveMemory(size);
+ if (availableOrReserved >= size) {
+ return;
+ }
+ } while (refprocActive);
+
+ // trigger VM's Reference processing
+ //noinspection CallToSystemGC
Review comment:
IntelliJ inspection
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586932050
<!--
Meta data
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Status:FAILURE URL:https://travis-ci.com/flink-ci/flink/builds/149255925 TriggerType:PUSH TriggerID:c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31
Hash:c71e8213f389b56ba642137c278c35272af3d58c Status:PENDING URL:https://travis-ci.com/flink-ci/flink/builds/149430663 TriggerType:PUSH TriggerID:c71e8213f389b56ba642137c278c35272af3d58c
-->
## CI report:
* c5a72c3274a7ba0edb6b4a6e1332cd3522d45e31 Travis: [FAILURE](https://travis-ci.com/flink-ci/flink/builds/149255925) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=5241)
* c71e8213f389b56ba642137c278c35272af3d58c Travis: [PENDING](https://travis-ci.com/flink-ci/flink/builds/149430663)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410253550
##########
File path: flink-core/src/main/java/org/apache/flink/util/JavaGcCleanerWrapper.java
##########
@@ -254,5 +339,31 @@ public Runnable create(Object owner, Runnable cleanupOperation) {
}
};
}
+
+ @Override
+ public boolean waitForGcToRunReadyCleaners() throws InterruptedException {
+ Object javaLangRefAccess = getJavaLangRefAccess();
+ try {
+ return (Boolean) waitForReferenceProcessingMethod.invoke(javaLangRefAccess);
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ if (e.getCause() instanceof InterruptedException) {
+ //noinspection ThrowInsideCatchBlockWhichIgnoresCaughtException
Review comment:
Please don't add IntelliJ specific inspection suppressions.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] azagrebin commented on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
azagrebin commented on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-587491324
> the NIO stack can only access the memory directly if it finds a DirectByteBuffer.
Ok then true, that workaround will not work
> Instead, they start with a big DirectByteBuffer (arena or chunk)
Ok, this might work. Just if I understand correctly, this will contribute to JVM direct memory limit then. I thought, in general we wanted to keep it only for external/user code. Depending on how much memory pre-allocated and then (if) (de)allocated more, it will interfere with the user code more or less.
I think `MemorySegments`, allocated with this pooling approach, can still benefit from the GC cleaner implemented in this PR, just the GC scheduled release operation will be a bit different.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410265250
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/UnsafeMemoryBudget.java
##########
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.memory;
+
+import org.apache.flink.util.JavaGcCleanerWrapper;
+
+import javax.annotation.Nonnegative;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Tracker of memory allocation and release within a custom limit.
+ *
+ * <p>This memory management in Flink uses the same approach as Java uses to allocate direct memory
+ * and release it upon GC but memory here can be also released directly with {@link #releaseMemory(long)}.
+ * The memory reservation {@link #reserveMemory(long)} tries firstly to run all phantom cleaners, created with
+ * {@link org.apache.flink.core.memory.MemoryUtils#createMemoryGcCleaner(Object, long, Runnable)},
+ * for objects which are ready to be garbage collected. If memory is still not available, it triggers GC and
+ * continues to process any ready cleaners making {@link #MAX_SLEEPS} attempts before throwing {@link OutOfMemoryError}.
+ */
+class UnsafeMemoryBudget {
+ // max. number of sleeps during try-reserving with exponentially
+ // increasing delay before throwing OutOfMemoryError:
+ // 1, 2, 4, 8, 16, 32, 64, 128, 256, 512 (total 1023 ms ~ 1 s)
+ // which means that OOME will be thrown after 0.5 s of trying
Review comment:
```suggestion
// which means that OOME will be thrown after 1 s of trying
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] tillrohrmann commented on a change in pull request #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
tillrohrmann commented on a change in pull request #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#discussion_r410258640
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/memory/MemoryManager.java
##########
@@ -240,7 +234,8 @@ public void allocatePages(
Set<MemorySegment> segmentsForOwner = currentSegmentsForOwner == null ?
new HashSet<>(numberOfPages) : currentSegmentsForOwner;
for (long i = numberOfPages; i > 0; i--) {
- MemorySegment segment = allocateOffHeapUnsafeMemory(getPageSize(), owner);
+ int size = getPageSize();
+ MemorySegment segment = allocateOffHeapUnsafeMemory(size, owner, () -> memoryBudget.releaseMemory(size));
Review comment:
If `getPageSize()` is constant, then I would suggest to create a single `() -> memoryBudget.releaseMemory(size)` instance. One could even have a method `releasePageSize() { memoryBudget.releaseMemory(getPageSize()); }` and then define outside of the loop a `Runnable releasePageSize = this::releasePageSize` and then pass `releasePageSize` to `allocateOffHeapUnsafeMemory`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11109:
[FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11109: [FLINK-15758][MemManager] Release segment and its unsafe memory in GC Cleaner
URL: https://github.com/apache/flink/pull/11109#issuecomment-586914869
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit c71e8213f389b56ba642137c278c35272af3d58c (Fri Feb 28 21:47:48 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services