This is an automated email from the ASF dual-hosted git repository.
rexxiong pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.3 by this push:
new a8665f125 [CELEBORN-1083] Refine TimeSlidingHub code
a8665f125 is described below
commit a8665f125ade9b0ddba062cd22b4bb5e420b1093
Author: onebox-li <[email protected]>
AuthorDate: Wed Oct 25 17:04:39 2023 +0800
[CELEBORN-1083] Refine TimeSlidingHub code
### What changes were proposed in this pull request?
Refine some TimeSlidingHub code.
### Why are the changes needed?
Ditto
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
UT.
Closes #2033 from onebox-li/refine.
Authored-by: onebox-li <[email protected]>
Signed-off-by: Shuang <[email protected]>
(cherry picked from commit bf42741ebc29e267c8c68f7a1cec785259a2dad1)
Signed-off-by: Shuang <[email protected]>
---
.../worker/congestcontrol/BufferStatusHub.java | 12 +++++--
.../worker/congestcontrol/TimeSlidingHub.java | 39 +++++++++++-----------
2 files changed, 29 insertions(+), 22 deletions(-)
diff --git
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/BufferStatusHub.java
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/BufferStatusHub.java
index 88c42bb52..c674673be 100644
---
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/BufferStatusHub.java
+++
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/BufferStatusHub.java
@@ -41,13 +41,19 @@ public class BufferStatusHub extends
TimeSlidingHub<BufferStatusHub.BufferStatus
@Override
public void combineNode(TimeSlidingNode node) {
BufferStatusNode needToCombined = (BufferStatusNode) node;
- numBytes.add(needToCombined.numBytes());
+ long nodeNumBytes = needToCombined.numBytes();
+ if (nodeNumBytes != 0) {
+ numBytes.add(nodeNumBytes);
+ }
}
@Override
public void separateNode(TimeSlidingNode node) {
- BufferStatusNode needToCombined = (BufferStatusNode) node;
- numBytes.add(-needToCombined.numBytes());
+ BufferStatusNode needToSeparate = (BufferStatusNode) node;
+ long nodeNumBytes = needToSeparate.numBytes();
+ if (nodeNumBytes != 0) {
+ numBytes.add(-nodeNumBytes);
+ }
}
@Override
diff --git
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/TimeSlidingHub.java
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/TimeSlidingHub.java
index 6951b0d5d..64ce466cc 100644
---
a/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/TimeSlidingHub.java
+++
b/worker/src/main/java/org/apache/celeborn/service/deploy/worker/congestcontrol/TimeSlidingHub.java
@@ -25,11 +25,11 @@ import org.apache.commons.lang3.tuple.Pair;
/**
* A time sliding list that group different {@link TimeSlidingNode} with
corresponding timestamp by
- * exact interval 1 second. Internally hold a {@link sumInfo} to get the sum
of the nodes in the
- * list.
+ * exact interval 1 second. Internally hold a {@link TimeSlidingHub#sumInfo}
to get the sum of the
+ * nodes in the list.
*
- * <p>This list is thread-safe, but {@link TimeSlidingNode} returned by the
method {@link sum}
- * should only be readable, do not try to update it.
+ * <p>This list is thread-safe, but {@link TimeSlidingNode} returned by the
method {@link
+ * TimeSlidingHub#sum()} should only be readable, do not try to update it.
*/
public abstract class TimeSlidingHub<N extends TimeSlidingHub.TimeSlidingNode>
{
@@ -44,7 +44,7 @@ public abstract class TimeSlidingHub<N extends
TimeSlidingHub.TimeSlidingNode> {
/** Merge new node with this. */
void combineNode(TimeSlidingNode node);
- /** Minus the value from the {@param node}. */
+ /** Minus the value from the node. */
void separateNode(TimeSlidingNode node);
TimeSlidingNode clone();
@@ -68,13 +68,12 @@ public abstract class TimeSlidingHub<N extends
TimeSlidingHub.TimeSlidingNode> {
}
public void add(N newNode) {
- long currentTimestamp = currentTimeMillis();
- add(currentTimestamp, newNode);
+ add(currentTimeMillis(), newNode);
}
public synchronized void add(long currentTimestamp, N newNode) {
- if (_deque.size() == 0) {
- _deque.add(Pair.of(currentTimestamp, (N) newNode.clone()));
+ if (sumInfo.getRight() == 0) {
+ _deque.add(Pair.of(currentTimestamp, newNode));
sumInfo = Pair.of((N) newNode.clone(), 1);
return;
}
@@ -92,7 +91,7 @@ public abstract class TimeSlidingHub<N extends
TimeSlidingHub.TimeSlidingNode> {
// The new node exceed existing sliding list, need to clear all old
nodes
// and create a new sliding list
_deque.clear();
- _deque.add(Pair.of(currentTimestamp, (N) newNode.clone()));
+ _deque.add(Pair.of(currentTimestamp, newNode));
sumInfo = Pair.of((N) newNode.clone(), 1);
return;
}
@@ -104,7 +103,7 @@ public abstract class TimeSlidingHub<N extends
TimeSlidingHub.TimeSlidingNode> {
_deque.add(lastNode);
}
- _deque.add(Pair.of(lastNode.getLeft() + intervalPerBucketInMills, (N)
newNode.clone()));
+ _deque.add(Pair.of(lastNode.getLeft() + intervalPerBucketInMills,
newNode));
N nodeToCombine = sumInfo.getLeft();
nodeToCombine.combineNode(newNode);
sumInfo = Pair.of(nodeToCombine, sumInfo.getRight() + nodesToAdd);
@@ -119,14 +118,16 @@ public abstract class TimeSlidingHub<N extends
TimeSlidingHub.TimeSlidingNode> {
}
if (timeDiff < 0) {
- // Belong to one existing node
- Iterator<Pair<Long, N>> iter = _deque.descendingIterator();
- while (iter.hasNext()) {
- Pair<Long, N> curNode = iter.next();
- if (currentTimestamp - curNode.getLeft() >= 0) {
- curNode.getRight().combineNode(newNode);
- sumInfo.getLeft().combineNode(newNode);
- return;
+ if (-timeDiff < intervalPerBucketInMills * (sumInfo.getRight() - 1)) {
+ // Belong to one existing node
+ Iterator<Pair<Long, N>> iter = _deque.descendingIterator();
+ while (iter.hasNext()) {
+ Pair<Long, N> curNode = iter.next();
+ if (currentTimestamp - curNode.getLeft() >= 0) {
+ curNode.getRight().combineNode(newNode);
+ sumInfo.getLeft().combineNode(newNode);
+ return;
+ }
}
}
