You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2023/06/17 08:51:54 UTC
[GitHub] [incubator-pekko] He-Pin opened a new pull request, #408: =str Make use of TireMap instead of List to reduce contention.
He-Pin opened a new pull request, #408:
URL: https://github.com/apache/incubator-pekko/pull/408
Was: https://github.com/akka/akka/pull/31262/
Use an ConcurrentMap as set to trace the contention in onFeedbackDispatched
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#issuecomment-1666823495
@pjfanning I have updated it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #408: =str Make use of ConucrrentHashMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285189907
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
The `AtomicRef` is needed as the callback can be called in a `AsyncCallback` which can be accessed from another thread.
Otherwise it will be a `@volatile asyncCallbacksInProgress` I think to indicate the stage has been stoped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on pull request #408: =str Make use of TireMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#issuecomment-1595683334
@mdedetrich I think this is a bug fix not a performance optimization, I can switch to `ConcurrentHashMap` i you like.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] pjfanning commented on a diff in pull request #408: =str Make use of ConucrrentHashMap instead of List to reduce contention.
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285185826
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
why use an AtomicReference on a mutable data structure like a ConcurrentHashMap? why not just mutate the map?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] pjfanning commented on a diff in pull request #408: =str Make use of ConucrrentHashMap instead of List to reduce contention.
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285193014
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
ConcurrentHashMaps are thread safe. I would suggest that we should look at making asyncCallbacksInProgress a val and we can do the usual Map calls on the Map - get, put, clear, etc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285198132
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
Noop, it will still create a map, I will update the pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] pjfanning commented on a diff in pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285195753
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
> That will not work, because an empty `ConcurrnetHashMap` can't indicate whether the stage is stopped. you can take a look at the comments upper 2~3 lines of this line.
Keep the AtomicReference then but could could to create the Set using this?
https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ConcurrentHashMap.html#newKeySet--
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #408: =str Make use of ConucrrentHashMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285194983
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
That will not work, because an empty `ConcurrnetHashMap` can't indicate whether the stage is stopped. you can take a look at the comments upper 2~3 lines of this line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on pull request #408: =str Make use of TrieMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#issuecomment-1666759706
After read the paper of TrieMap, I think I need to change it back to ConcurrentHashMap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin merged pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] jrudolph commented on pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "jrudolph (via GitHub)" <gi...@apache.org>.
jrudolph commented on PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#issuecomment-1674413926
Yes, this was originally a performance improvement. Regardless of that fact, let's not just merge changes in performance critical code without running benchmarks or discussing arguments why they can be omitted.
It has been a long time since I worked on this / proposed the change (almost 3 years... oh my). I guess in this case, it's fine, it seems after all they ran some benchmarks in Akka to validate it does not introduce obvious performance regressions.
> I'm a bit confused as to the use of `AtomicReference[mutable.Map]` instead of just plain `mutable.Map`.
The comment at its declaration explains it. There's a race condition between scheduling an async callback (i.e. an incoming external event into the stream) and shutting down a stream. The atomic reference makes sure, that we will report feedback to the async callback reliably even if it raced against shutdown (this is basically what this whole feature is about).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] mdedetrich commented on pull request #408: =str Make use of TrieMap instead of List to reduce contention.
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#issuecomment-1595693341
Ah I see, I will get @jrudolph to review this since he made the original issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #408: =str Make use of ConucrrentHashMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285190258
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
@pjfanning do you suggest I chang it to `var @volatile asyncCallbacksInProgress` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] pjfanning commented on a diff in pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285197531
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
This saves use from having to create a map and then create a keyset - there is just one call to create a keyset.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285199494
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
@pjfanning I have updated as you suggestion.
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
@pjfanning I have updated as your suggestion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] mdedetrich commented on pull request #408: =str Make use of TireMap instead of List to reduce contention.
Posted by "mdedetrich (via GitHub)" <gi...@apache.org>.
mdedetrich commented on PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#issuecomment-1595679374
Is it possible to provide jmh benchmarks to show the difference?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] pjfanning commented on a diff in pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285197348
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
Does it matter what the value type is? The Map needs a value type but you are only interested in the key type.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
[GitHub] [incubator-pekko] He-Pin commented on a diff in pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #408:
URL: https://github.com/apache/incubator-pekko/pull/408#discussion_r1285196630
##########
stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala:
##########
@@ -1282,7 +1283,9 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
private var callbacksWaitingForInterpreter: List[ConcurrentAsyncCallback[_]] = Nil
// is used for two purposes: keep track of running callbacks and signal that the
// stage has stopped to fail incoming async callback invocations by being set to null
- private val asyncCallbacksInProgress = new AtomicReference[List[Promise[Done]]](Nil)
+ // Using ConcurrentHashMap's KeySetView as Set to track the inProgress async callbacks.
+ private val asyncCallbacksInProgress: AtomicReference[java.util.Set[Promise[Done]]] =
+ new AtomicReference(new ConcurrentHashMap[Promise[Done], NotUsed]().keySet(NotUsed))
Review Comment:
But the default value is BOOLEAN.TRUE, seems a little weird in this case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org