[#825][followup] fix(spark): Apply a thread safety way to track the blocks sending result - #1260
Conversation
|
@yl09099 @jerqi PTAL |
| blockId -> | ||
| blockIdsSendFailTracker | ||
| .computeIfAbsent(blockId, id -> Lists.newArrayList()) | ||
| .computeIfAbsent(blockId, id -> new CopyOnWriteArrayList()) |
There was a problem hiding this comment.
CopyOnWriteArrayList is suitable for the situation that we have more read operations than write operations. But we have more write operations than read operations now. Maybe it will give us performance regression. If we want to use thread safe collection, maybe queue will be better.
There was a problem hiding this comment.
Ur right, replace it with BlockingQueue. Besides I revert the blockIdsSendSuccessTracker to the earlier code, since we don't track the success server.
summaryzb
changed the title
Codecov Report
@@ Coverage Diff @@
## master #1260 +/- ##
============================================
+ Coverage 53.93% 55.06% +1.13%
- Complexity 2647 2653 +6
============================================
Files 396 376 -20
Lines 23036 20687 -2349
Branches 1970 1970
============================================
- Hits 12424 11391 -1033
+ Misses 9845 8600 -1245
+ Partials 767 696 -71
... and 21 files with indirect coverage changes 📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more |
| // we assume that most of the blocks can be sent successfully | ||
| // so initialize the map at first without concurrency insurance | ||
| // AtomicInteger is enough to reflect value changes in other threads | ||
| Map<Long, AtomicInteger> blockIdsSendSuccessTracker = Maps.newHashMap(); |
There was a problem hiding this comment.
Initialize With all blockIds to reduce the condition race, eliminate the scenario that no existing blockId while tracking the sending result
| // we assume that most of the blocks can be sent successfully | ||
| // so initialize the map at first without concurrency insurance | ||
| // AtomicInteger is enough to reflect value changes in other threads | ||
| Map<Long, AtomicInteger> blockIdsSendSuccessTracker = Maps.newHashMap(); |
There was a problem hiding this comment.
BlockIdsSendSuccessTracker object should have concurrency issues? Use concurrentMap?
There was a problem hiding this comment.
No concurrency issues, since blockIdsSendSuccessTracker is already initialize with all blockIds, it will not be modified in other thread.
There was a problem hiding this comment.
Instead of put the blockId in blockIdsSendSuccessTracker when get a response in multi threads, We just put all blockIds in the beginning before concurrency execute. This is helpful to reduce condition race, and we can be optimistically that no space will be wasted since most of the blockIds will be sent successfully.
roryqi
left a comment
roryqi
left a comment
There was a problem hiding this comment.
LGTM, thanks @summaryzb @yl09099
roryqi
changed the title
4 participants
What changes were proposed in this pull request?
As title
Why are the changes needed?
ActionLink
I debug the local test and find that all blocks are successfully send, but some blocks are not in the block tracker
Does this PR introduce any user-facing change?
No.
How was this patch tested?
Integration test
Especially run below test in a loop of many times without a fail