Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[#2197] Support reg app conf to server and avoid update committed/cached blockIds bitmap #2196

Merged
merged 6 commits into from
Oct 29, 2024
Merged

[#2197] Support reg app conf to server and avoid update committed/cached blockIds bitmap #2196

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
8c266eb
Support reg app conf to server and avoid update committedBlocksId and…
maobaolong Oct 15, 2024
fb83e78
Address comment, rename config key
maobaolong Oct 28, 2024
dcb37e3
Address suggestion and use server side configured storage type
maobaolong Oct 28, 2024
df22106
Address suggestion check commit/finish shuffle
maobaolong Oct 28, 2024
7317891
Address comments
maobaolong Oct 29, 2024
714ef98
fix ut
maobaolong Oct 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 1 addition & 3 deletions server/src/main/java/org/apache/uniffle/server/ShuffleFlushManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,9 +103,7 @@ public ShuffleFlushManager(
.mapToLong(bitmap -> bitmap.getLongCardinality())
.sum(),
2 * 60 * 1000L /* 2 minutes */);
this.storageTypeWithMemory =
StorageType.withMemory(
StorageType.valueOf(shuffleServerConf.get(ShuffleServerConf.RSS_STORAGE_TYPE).name()));
this.storageTypeWithMemory = StorageType.withMemory(StorageType.valueOf(storageType));
}

public void addToFlushQueue(ShuffleDataFlushEvent event) {
Expand Down
19 changes: 19 additions & 0 deletions server/src/main/java/org/apache/uniffle/server/ShuffleServerGrpcService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,7 @@
import org.apache.uniffle.storage.common.Storage;
import org.apache.uniffle.storage.common.StorageReadMetrics;
import org.apache.uniffle.storage.util.ShuffleStorageUtils;
import org.apache.uniffle.storage.util.StorageType;

import static org.apache.uniffle.server.merge.ShuffleMergeManager.MERGE_APP_SUFFIX;

Expand Down Expand Up @@ -578,6 +579,15 @@ public void commitShuffleTask(
String appId = req.getAppId();
int shuffleId = req.getShuffleId();
auditContext.withAppId(appId).withShuffleId(shuffleId);
org.apache.uniffle.common.StorageType storageType =
shuffleServer.getShuffleServerConf().get(ShuffleServerConf.RSS_STORAGE_TYPE);
boolean storageTypeWithMemory =
StorageType.withMemory(StorageType.valueOf(storageType.name()));
if (storageTypeWithMemory) {
LOG.warn(
"finishShuffle should not be called while server-side configured StorageType to {}",
storageType);
}
StatusCode status = verifyRequest(appId);
if (status != StatusCode.SUCCESS) {
auditContext.withStatusCode(status);
Expand Down Expand Up @@ -635,6 +645,15 @@ public void finishShuffle(
String appId = req.getAppId();
int shuffleId = req.getShuffleId();
auditContext.withAppId(appId).withShuffleId(shuffleId);
org.apache.uniffle.common.StorageType storageType =
shuffleServer.getShuffleServerConf().get(ShuffleServerConf.RSS_STORAGE_TYPE);
boolean storageTypeWithMemory =
StorageType.withMemory(StorageType.valueOf(storageType.name()));
if (storageTypeWithMemory) {
LOG.warn(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should throw exception directly.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@jerqi If we throw exception, it will never support for test mode and LOCALFILE client to access the server with LOCALFILE_MEMORY storageType?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK for me.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I means that it's ok for me to throw exception here.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@jerqi Sorry for the misunderstand, done.

"finishShuffle should not be called while server-side configured StorageType to {}",
storageType);
}
StatusCode status = verifyRequest(appId);
if (status != StatusCode.SUCCESS) {
auditContext.withStatusCode(status);
Expand Down
7 changes: 5 additions & 2 deletions server/src/main/java/org/apache/uniffle/server/ShuffleTaskManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,9 +149,12 @@ public ShuffleTaskManager(
this.shuffleBufferManager = shuffleBufferManager;
this.storageManager = storageManager;
this.shuffleMergeManager = shuffleMergeManager;
org.apache.uniffle.common.StorageType storageType =
conf.get(ShuffleServerConf.RSS_STORAGE_TYPE);
this.storageTypeWithMemory =
StorageType.withMemory(
StorageType.valueOf(conf.get(ShuffleServerConf.RSS_STORAGE_TYPE).name()));
storageType == null
? false
: StorageType.withMemory(StorageType.valueOf(storageType.name()));
this.appExpiredWithoutHB = conf.getLong(ShuffleServerConf.SERVER_APP_EXPIRED_WITHOUT_HEARTBEAT);
this.commitCheckIntervalMax = conf.getLong(ShuffleServerConf.SERVER_COMMIT_CHECK_INTERVAL_MAX);
this.preAllocationExpired = conf.getLong(ShuffleServerConf.SERVER_PRE_ALLOCATION_EXPIRED);
Expand Down
Loading