Github user aarondav commented on a diff in the pull request:
https://github.com/apache/spark/pull/5743#discussion_r29453868
--- Diff:
network/common/src/main/java/org/apache/spark/network/server/OneForOneStreamManager.java
---
@@ -56,6 +62,15 @@ public OneForOneStreamManager() {
// This does not need to be globally unique, only unique to this class.
nextStreamId = new AtomicLong((long) new
Random().nextInt(Integer.MAX_VALUE) * 1000);
streams = new ConcurrentHashMap<Long, StreamState>();
+ streamIds = new ConcurrentHashMap<Channel, Set<Long>>();
+ }
+
+ @Override
+ public void registerChannel(Channel channel, long streamId) {
--- End diff --
I think we can avoid the new field by doing
```java
streams.get(streamId).associatedChannel = channel;
```
here and
```java
val streamIterator = streams.iterator()
while (streamIterator.hasNext()) {
StreamState state = streamIterator.next().getValue()
if (state.associatedChannel == channel) {
streamIterator.remove();
// Release all remaining buffers.
while (state.buffers.hasNext()) {
state.buffers.next().release();
}
}
}
```
Allowing the removal of the other connectionTerminated().
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
