Skip to content

[SPARK-30406] OneForOneStreamManager ensure that compound operations on shared variables are atomic #27071

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

Closed
ajithme wants to merge 2 commits into from
Closed

[SPARK-30406] OneForOneStreamManager ensure that compound operations on shared variables are atomic #27071

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9be777c
Using compound operations as well as increments and decrements on pri…
ajithme Jan 2, 2020
48c9058
Make final
ajithme Jan 2, 2020
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
8 changes: 4 additions & 4 deletions .../network-common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ private static class StreamState {
int curChunk = 0;

// Used to keep track of the number of chunks being transferred and not finished yet.
volatile long chunksBeingTransferred = 0L;
final AtomicLong chunksBeingTransferred = new AtomicLong(0L);

StreamState(String appId, Iterator<ManagedBuffer> buffers, Channel channel) {
this.appId = appId;
Expand Down Expand Up @@ -153,7 +153,7 @@ public void checkAuthorization(TransportClient client, long streamId) {
public void chunkBeingSent(long streamId) {
StreamState streamState = streams.get(streamId);
if (streamState != null) {
streamState.chunksBeingTransferred++;
streamState.chunksBeingTransferred.incrementAndGet();
}

}
Expand All @@ -167,7 +167,7 @@ public void streamBeingSent(String streamId) {
public void chunkSent(long streamId) {
StreamState streamState = streams.get(streamId);
if (streamState != null) {
streamState.chunksBeingTransferred--;
streamState.chunksBeingTransferred.decrementAndGet();
}
}

Expand All @@ -180,7 +180,7 @@ public void streamSent(String streamId) {
public long chunksBeingTransferred() {
long sum = 0L;
for (StreamState streamState: streams.values()) {
sum += streamState.chunksBeingTransferred;
sum += streamState.chunksBeingTransferred.get();
}
return sum;
}
Expand Down