[GitHub] [incubator-pekko] pjfanning commented on a diff in pull request #408: =str Make use of ConcurrentHashMap instead of List to reduce contention.

["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