iframe-proxy

rickyma · 2024-02-10T20:37:44Z

What changes were proposed in this pull request?

When the shuffle server enables Netty, during the pre-allocation of memory and flushing buffer, we should use the actual used direct memory(which is pinnedDirectMemory in PooledByteBufAllocator) and maxDirectMemory for the if statement, instead of the previous usedMemory and capacity due to #1472.

Changes:

When pre-allocating memory and flushing buffer, use the actual direct memory as the basis for judgment, instead of relying on the previous usedMemory metric.
No need to allocate a new direct ByteBuf in Decoders.decodeShuffleBlockInfo(), we can reuse it using readSlice.
Add two metrics allocated_direct_memory_size and pinned_direct_memory_size. Refer to Add pinnedHeap/DirectMemory methods to ByteBufAllocatorMetric netty/netty#11667.
Improve the robustness of methods ShuffleDataResult.release() and ShuffleIndexResult.release() to fix occasional IllegalReferenceCountException issues in extremely rare scenarios.
Upgrade Netty to latest stable version: 4.1.106.final (including a lot of bug fixes and an important feature pinnedDirectMemory which is needed in this PR).
Upgrade GRPC to latest stable version: 1.61.1 (including a lot of bug fixes).
Fix #1008. It does not actually test GRPC_NETTY mode, because it uses ShuffleServerGrpcClient everywhere instead of ShuffleServerGrpcNettyClient. (I found this problem when fixing the previous UTs. Setting the shuffle server's tags to GRPC_NETTY,GRPC is useless, because we are not using ShuffleServerGrpcNettyClient at all.)

Why are the changes needed?

For #1472

Detailed explanations：
The actual direct memory usage of a ByteBuf that is going to be sent to a Netty server cannot be obtained in advance due to the dynamic nature of Netty's allocation and deallocation.

We use a pooled ByteBufAllocator by default for performance reasons. When a ByteBuf is allocated, it is drawn from a pool of available memory, and when it is deallocated, it is returned to the pool. The size of the ByteBuf can change dynamically as data is written to or read from it.

Furthermore, the actual memory usage includes not only the capacity of the ByteBuf, but also additional overhead for the ByteBuf object itself and possibly other associated objects. This overhead is not constant and can vary depending on factors such as the JVM implementation, the specific Netty version, the runtime environment, and even the network.
Therefore, it is generally not possible to accurately predict the actual direct memory usage of a ByteBuf in advance.

While we were using it, we discovered that due to Netty's dynamic allocation of direct memory based on page and chunk sizes, the actual direct memory occupied by the shuffle server gradually deviates from usedMemory over time. Consequently, it fails to meet the flush conditions for an extended period, as flushing or not relies on usedMemory. This results in direct used memory exceeding its limit, which explains why an out-of-direct-memory error is prone to occur under high-pressure and high-concurrency situations. If the pressure is insufficient, the job might be completed before this issue arises.

Does this PR introduce any user-facing change?

In GRPC_NETTY mode, usedMemory will use pinnedDirectMemory.
Default values of rss.server.netty.directMemoryTracker.memoryUsage.updateMetricsIntervalMs and rss.server.netty.directMemoryTracker.memoryUsage.initialFetchDelayMs configurations are decreased to 1s.

How was this patch tested?

Fix #1008 to make it test GRPC_NETTY mode using ShuffleServerGrpcNettyClient.
We conducted stress tests in our production environment and the spark job ran successfully.
Before this PR, the spark job cannot run successfully.

…erver OOM when enabling Netty

rickyma · 2024-02-10T20:47:40Z

codecov-commenter · 2024-02-10T21:28:48Z

Codecov Report

Attention: 36 lines in your changes are missing coverage. Please review.

Comparison is base (576a925) 54.27% compared to head (9ace0b3) 54.57%.
Report is 8 commits behind head on master.

Files	Patch %	Lines
...he/uniffle/server/buffer/ShuffleBufferManager.java	58.82%	11 Missing and 10 partials ⚠️
...a/org/apache/uniffle/common/ShuffleServerInfo.java	0.00%	6 Missing ⚠️
...pache/uniffle/server/NettyDirectMemoryTracker.java	50.00%	4 Missing ⚠️
.../org/apache/uniffle/common/ShuffleIndexResult.java	0.00%	2 Missing ⚠️
...a/org/apache/uniffle/common/ShuffleDataResult.java	0.00%	1 Missing ⚠️
...ava/org/apache/uniffle/common/util/NettyUtils.java	50.00%	1 Missing ⚠️
...pache/uniffle/server/ShuffleServerGrpcService.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #1519      +/-   ##
============================================
+ Coverage     54.27%   54.57%   +0.30%     
+ Complexity     2807     2791      -16     
============================================
  Files           427      410      -17     
  Lines         24349    22123    -2226     
  Branches       2077     2100      +23     
============================================
- Hits          13215    12074    -1141     
+ Misses        10305     9290    -1015     
+ Partials        829      759      -70

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

github-actions · 2024-02-10T21:57:29Z

Test Results

2 042 files - 245 2 042 suites - 245 2h 20m 15s ⏱️ - 2h 9m 28s
753 tests - 66 750 ✅ - 68 1 💤 ±0 0 ❌ ±0 2 🔥 + 2
9 172 runs + 86 9 130 ✅ + 57 14 💤 +1 0 ❌ ±0 28 🔥 +28

For more details on these errors, see this check.

Results for commit d890a53. ± Comparison against base commit 3954ce7.

This pull request removes 73 and adds 7 tests. Note that renamed tests count towards both.

org.apache.uniffle.server.TopNShuffleDataSizeOfAppCalcTaskTest ‑ testTopNShuffleDataSizeOfAppCalcTask
org.apache.uniffle.test.AQERepartitionTest ‑ resultCompareTest
org.apache.uniffle.test.AQESkewedJoinTest ‑ resultCompareTest
org.apache.uniffle.test.AQESkewedJoinWithLocalOrderTest ‑ resultCompareTest
org.apache.uniffle.test.AutoAccessTest ‑ test
org.apache.uniffle.test.CombineByKeyTest ‑ combineByKeyTest
org.apache.uniffle.test.ContinuousSelectPartitionStrategyTest ‑ resultCompareTest
org.apache.uniffle.test.DiskErrorToleranceTest ‑ diskErrorTest
org.apache.uniffle.test.DynamicConfTest ‑ dynamicConfTest
org.apache.uniffle.test.DynamicFetchClientConfTest ‑ test
…

org.apache.uniffle.test.ShuffleServerConcurrentWriteOfHadoopTest ‑ testConcurrentWrite2Hadoop{int, int, boolean}[1]
org.apache.uniffle.test.ShuffleServerConcurrentWriteOfHadoopTest ‑ testConcurrentWrite2Hadoop{int, int, boolean}[2]
org.apache.uniffle.test.ShuffleServerConcurrentWriteOfHadoopTest ‑ testConcurrentWrite2Hadoop{int, int, boolean}[3]
org.apache.uniffle.test.ShuffleServerConcurrentWriteOfHadoopTest ‑ testConcurrentWrite2Hadoop{int, int, boolean}[4]
org.apache.uniffle.test.ShuffleServerOnRandomPortTest ‑ org.apache.uniffle.test.ShuffleServerOnRandomPortTest
org.apache.uniffle.test.ShuffleServerWithMemLocalHadoopTest ‑ memoryLocalFileHadoopReadWithFilterTest{boolean, boolean}[1]
org.apache.uniffle.test.ShuffleServerWithMemLocalHadoopTest ‑ memoryLocalFileHadoopReadWithFilterTest{boolean, boolean}[2]

♻️ This comment has been updated with latest results.

roryqi · 2024-02-12T02:13:19Z

Maybe we can spit this pull requests to some smaller pull requests. It will be easier to review. Surely, it's good to summit a drat pull request at request to give reviewers an overall view.

rickyma · 2024-02-12T06:04:18Z

Maybe we can spit this pull requests to some smaller pull requests. It will be easier to review. Surely, it's good to summit a drat pull request at request to give reviewers an overall view.

Smaller pull requests will be:

Upgrade Netty to latest stable version: 4.1.106.final.
Upgrade GRPC to latest stable version: 1.61.1 (including a lot of bug fixes).
No need to allocate a new direct ByteBuf in Decoders.decodeShuffleBlockInfo(), we can reuse it using readSlice.
Improve the robustness of methods ShuffleDataResult.release() and ShuffleIndexResult.release() to fix occasional IllegalReferenceCountException issues in extremely rare scenarios.
Add two metrics allocated_direct_memory_size and pinned_direct_memory_size.
6.1 and 6.2 should be in one pull request, or else the UTs will fail.
6.1 When pre-allocating memory and flushing buffer, use the actual direct memory as the basis for judgment, instead of relying on the previous usedMemory metric.
6.2 Fix #1008. It does not actually test GRPC_NETTY mode, because it uses ShuffleServerGrpcClient everywhere instead of ShuffleServerGrpcNettyClient.

Maybe 1 and 5 can be in one pull request too?

WDYT? @jerqi

roryqi · 2024-02-12T06:31:00Z

Maybe we can spit this pull requests to some smaller pull requests. It will be easier to review. Surely, it's good to summit a drat pull request at request to give reviewers an overall view.

Smaller pull requests will be:

Upgrade Netty to latest stable version: 4.1.106.final.

Upgrade GRPC to latest stable version: 1.61.1 (including a lot of bug fixes).

No need to allocate a new direct ByteBuf in Decoders.decodeShuffleBlockInfo(), we can reuse it using readSlice.

Improve the robustness of methods ShuffleDataResult.release() and ShuffleIndexResult.release() to fix occasional IllegalReferenceCountException issues in extremely rare scenarios.

Add two metrics allocated_direct_memory_size and pinned_direct_memory_size.

6.1 and 6.2 should be in one pull request, or else the UTs will fail.
6.1 When pre-allocating memory and flushing buffer, use the actual direct memory as the basis for judgment, instead of relying on the previous usedMemory metric.
6.2 Fix #1008. It does not actually test GRPC_NETTY mode, because it uses ShuffleServerGrpcClient everywhere instead of ShuffleServerGrpcNettyClient.

Maybe 1 and 5 can be in one pull request too?

WDYT? @jerqi

1 and 2 can be one pull request, too. Five pull requests seems easier to review.

rickyma · 2024-02-12T07:41:06Z

I'll initially submit these three pull requests, and the remaining ones can be submitted once these are merged.
@jerqi

### What changes were proposed in this pull request? Upgrade Netty and GRPC ### Why are the changes needed? A sub PR for: #1519 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs.

…Memory and usedDirectMemory (#1524) ### What changes were proposed in this pull request? We need to know the exact direct memory usage of `PooledByteBufAllocator` in `Netty`. So we should introduce metrics for Netty's `pinnedDirectMemory` and `usedDirectMemory`. ### Why are the changes needed? A sub PR for: #1519 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs.

…ption issues in extremely rare scenarios (#1522) ### What changes were proposed in this pull request? Improve the robustness of methods `ShuffleDataResult.release()` and `ShuffleIndexResult.release()` to fix occasional IllegalReferenceCountException issues in extremely rare scenarios. ### Why are the changes needed? A sub PR for: #1519 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs.

…ks instead of reallocating it (#1521) ### What changes were proposed in this pull request? Reuse ByteBuf when decoding shuffle blocks instead of reallocating it ### Why are the changes needed? A sub PR for: #1519 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs.

zuston · 2024-02-18T02:28:38Z

I still don't understand the reason of OOM for this. Could you help explain why the usedMemory can't reflect the real memory used compared with grpc based server? @rickyma

rickyma · 2024-02-18T02:49:12Z

I still don't understand the reason of OOM for this. Could you help explain why the usedMemory can't reflect the real memory used compared with grpc based server? @rickyma

As described in this PR previously, Netty allocates direct memory by pages and chunks.

Furthermore, the actual memory usage includes not only the capacity of the ByteBuf, but also additional overhead for the ByteBuf object itself and possibly other associated objects. This overhead is not constant and can vary depending on factors such as the JVM implementation, the specific Netty version, the runtime environment, and even the network.
Therefore, it is generally not possible to accurately predict the actual direct memory usage of a ByteBuf in advance.

When the concurrent pressure of the shuffle server is not high, the ByteBuf allocated by Netty will be flushed and released in time, and there will be no problem. However, when the concurrent pressure is high, if it has not been flushed in time, the direct memory occupied by all ByteBuf allocated by Netty will be much larger than usedMemory. This is because, for example, the size of NORMAL-sized ByteBuf in PooledByteBufAllocator will be rounded up according to chunks and pages (
In reality, it would be more complicated, but for now, it can be simply understood this way), and each ByteBuf also has additional overhead in the Netty framework internally, causing the actual occupied direct memory size to gradually deviate from usedMemory over time, becoming larger and larger. When an OOM occurs, the value of usedMemory is still very small. When judging pre-allocation, it will still meet the conditions, allowing new sendShuffleDataRequest requests to allocate new memory, but at this time, it still does not meet the flush conditions because whether to flush or not is also determined based on usedMemory.

I don't know if it is clear to you now.

@zuston

zuston · 2024-02-18T03:16:37Z

the ByteBuf allocated by Netty will be flushed and released in time

Who will release and flush ByteBuf ?

rickyma · 2024-02-18T03:19:09Z

the ByteBuf allocated by Netty will be flushed and released in time

I mean flushSingleBufferIfNecessary and flushIfNecessary in ShuffleBufferManager.

zuston · 2024-02-18T05:45:15Z

the ByteBuf allocated by Netty will be flushed and released in time

I mean flushSingleBufferIfNecessary and flushIfNecessary in ShuffleBufferManager.

Got you point, so you suspect the bytebuf occupied memory size is not equals to the real used size, right? How about initializing the bytebuf with capacity like: ByteBuf buffer = Unpooled.buffer(1024);

rickyma · 2024-02-18T06:28:24Z

the ByteBuf allocated by Netty will be flushed and released in time

I mean flushSingleBufferIfNecessary and flushIfNecessary in ShuffleBufferManager.

Got you point, so you suspect the bytebuf occupied memory size is not equals to the real used size, right? How about initializing the bytebuf with capacity like: ByteBuf buffer = Unpooled.buffer(1024);

You can't precisely control the size of the ByteBuf received by Netty's channelRead, it may be arbitrarily truncated, and may even be affected by network fluctuations. To give a simple example, you send a 15MB ByteBuf, but due to network issues, the server might first receive a 10MB ByteBuf, and then a 5MB ByteBuf. Assuming the chunkSize is 16MB, it will most likely occupy two 16MB chunks of memory on the server, which is 32MB, and the Netty framework will also use some internal memory overhead, so the final memory usage will most likely be more than 32MB.

zuston · 2024-02-18T07:23:47Z

the ByteBuf allocated by Netty will be flushed and released in time

I mean flushSingleBufferIfNecessary and flushIfNecessary in ShuffleBufferManager.

Got you point, so you suspect the bytebuf occupied memory size is not equals to the real used size, right? How about initializing the bytebuf with capacity like: ByteBuf buffer = Unpooled.buffer(1024);

You can't precisely control the size of the ByteBuf received by Netty's channelRead, it may be arbitrarily truncated, and may even be affected by network fluctuations. To give a simple example, you send a 15MB ByteBuf, but due to network issues, the server might first receive a 10MB ByteBuf, and then a 5MB ByteBuf. Assuming the chunkSize is 16MB, it will most likely occupy two 16MB chunks of memory on the server, which is 32MB, and the Netty framework will also use some internal memory overhead, so the final memory usage will most likely be more than 32MB.

If so, why not creating a new fixed size bytebuf to fill above composite bytebuf? And the overhead memory occupied exceeded can be reflected to the client side using backpressure?

rickyma · 2024-02-18T07:46:34Z

If so, why not creating a new fixed size bytebuf to fill above composite bytebuf?

As I mentioned above, you cannot control the size of the ByteBuf received by the method channelRead.
When the channelRead method is executed, the ByteBuf has already occupied the direct memory of the shuffle server. At this point, no matter what method you use, this part of the memory will always be occupied. If you want to receive and concatenate them through a CompositeByteBuf, this will lead to problems similar to #1521. You have to create a new CompositeByteBuf, and then release the ByteBuf passed in by channelRead.

This means, for a very short time, the direct memory usage is doubled, which is extremely uncontrollable.

And it will cause the issue that I mentioned in #1521-comment.

rickyma · 2024-02-18T07:56:37Z

And the overhead memory occupied exceeded can be reflected to the client side using backpressure?

I think backpressure can be implemented in another PR, which would be a further improvement to this issue. The current PR is meant to address the problem of inaccurate usedMemory causing OOM.

…mory issue causing OOM (#1534) ### What changes were proposed in this pull request? When we use `UnpooledByteBufAllocator` to allocate off-heap `ByteBuf`, Netty directly requests off-heap memory from the operating system instead of allocating it according to `pageSize` and `chunkSize`. This way, we can obtain the exact `ByteBuf` size during the pre-allocation of memory, avoiding distortion of metrics such as `usedMemory`. Moreover, we have restored the code submission of the PR [#1521](#1521). We ensure that there is sufficient direct memory for the Netty server during decoding `sendShuffleDataRequest` by taking into account the `encodedLength` of `ByteBuf` in advance during the pre-allocation of memory, thus avoiding OOM during decoding `sendShuffleDataRequest`. Since we are not using `PooledByteBufAllocator`, the PR [#1524](#1524) is no longer needed. ### Why are the changes needed? A sub PR for: #1519 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs.

roryqi · 2024-02-26T01:58:17Z

### What changes were proposed in this pull request? Fix [#1008](#1008). It does not actually test `GRPC_NETTY` mode, because it uses `ShuffleServerGrpcClient` everywhere instead of `ShuffleServerGrpcNettyClient`. Setting the shuffle server's tags to `GRPC_NETTY,GRPC` is useless, because we are not using `ShuffleServerGrpcNettyClient` at all. ### Why are the changes needed? It is a sub PR for: #1519 Also, it is a follow-up PR for: #1008 ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing UTs.

rickyma added 5 commits February 5, 2024 13:51

[apache#1472] fix(server): Inaccurate flow control leads to Shuffle s…

89d2ae8

…erver OOM when enabling Netty

[apache#1472] fix(server): Inaccurate flow control leads to Shuffle s…

947bbf3

…erver OOM when enabling Netty

[apache#1472] fix(server): Inaccurate flow control leads to Shuffle s…

6980ae0

…erver OOM when enabling Netty

fix previous tests

a34c300

fix previous tests

a4db16e

rickyma force-pushed the issue-1472 branch 2 times, most recently from 6b8c440 to dc50ca4 Compare February 10, 2024 21:18

fix previous tests

a1600fa

rickyma force-pushed the issue-1472 branch from dc50ca4 to a1600fa Compare February 11, 2024 15:07

rickyma added 2 commits February 11, 2024 23:18

fix previous tests

3fb2c8f

Upgrade GRPC to latest

409a307

rickyma mentioned this pull request Feb 13, 2024

[#1472][part-4] feature(server): Introduce metrics for Netty's pinnedDirectMemory and usedDirectMemory #1524

Merged

rickyma force-pushed the issue-1472 branch from 9ace0b3 to 409a307 Compare February 14, 2024 07:52

Merge branch 'master' into issue-1472

988fc8b

rickyma force-pushed the issue-1472 branch from 94ba9c0 to 988fc8b Compare February 14, 2024 09:17

Fix usedMemory

d890a53

rickyma mentioned this pull request Feb 15, 2024

[#1472][part-5] Inaccurate flow control leads to Shuffle server OOM when enabling Netty #1531

Closed

rickyma mentioned this pull request Feb 21, 2024

[#1472][part-5] Use UnpooledByteBufAllocator to obtain accurate ByteBuf sizes to fix inaccurate usedMemory issue causing OOM #1534

Merged

rickyma mentioned this pull request Feb 24, 2024

[#1472][part-6] fix(netty): Make UTs truly test Netty mode #1540

Merged

roryqi closed this Feb 26, 2024

rickyma deleted the issue-1472 branch May 5, 2024 08:34

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

Conversation

rickyma commented Feb 10, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

rickyma commented Feb 10, 2024

Uh oh!

codecov-commenter commented Feb 10, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

github-actions Bot commented Feb 10, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Test Results

Uh oh!

roryqi commented Feb 12, 2024

Uh oh!

rickyma commented Feb 12, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

roryqi commented Feb 12, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rickyma commented Feb 12, 2024

Uh oh!

zuston commented Feb 18, 2024

Uh oh!

rickyma commented Feb 18, 2024

Uh oh!

zuston commented Feb 18, 2024

Uh oh!

rickyma commented Feb 18, 2024

Uh oh!

zuston commented Feb 18, 2024

Uh oh!

rickyma commented Feb 18, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

zuston commented Feb 18, 2024

Uh oh!

rickyma commented Feb 18, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

rickyma commented Feb 18, 2024

Uh oh!

roryqi commented Feb 26, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

rickyma commented Feb 10, 2024 •

edited

Loading

codecov-commenter commented Feb 10, 2024 •

edited

Loading

github-actions Bot commented Feb 10, 2024 •

edited

Loading

rickyma commented Feb 12, 2024 •

edited

Loading

roryqi commented Feb 12, 2024 •

edited

Loading

rickyma commented Feb 18, 2024 •

edited

Loading

rickyma commented Feb 18, 2024 •

edited

Loading