replace concurrently with labeled variant

Author: ffakenz

hydra-cluster/bench/Bench/EndToEnd.hs

@@ -386,9 +386,12 @@ processTransactions clients clientDatasets = do
     submissionQ <- newLabelledTBQueueIO "submission" (fromIntegral numberOfTxs)
     registry <- newRegistry
     atomically $ forM_ txSequence $ writeTBQueue submissionQ
-    ( submitTxs client registry submissionQ
-        `concurrently_` waitForAllConfirmations client registry (Set.fromList $ map txId txSequence)
-        `concurrently_` progressReport (hydraNodeId client) clientId numberOfTxs submissionQ
+    concurrentlyLabelled_
+      ("submit-txs", submitTxs client registry submissionQ)
+      ( "confirm-txs"
+      , concurrentlyLabelled_
+          ("wait-for-all-confirmations", waitForAllConfirmations client registry (Set.fromList $ map txId txSequence))
+          ("progress-report", progressReport (hydraNodeId client) clientId numberOfTxs submissionQ)
       )
       `catch` \(HUnitFailure sourceLocation reason) ->
         putStrLn ("Something went wrong while waiting for all confirmations: " <> formatLocation sourceLocation <> ": " <> formatFailureReason reason)

hydra-cluster/test/Test/EndToEndSpec.hs

@@ -676,9 +676,9 @@ spec = around (showLogsOnFailure "EndToEndSpec") $ do
           withClusterTempDir $ \tmpDir -> do
             withBackend (contramap FromCardanoNode tracer) tmpDir $ \_ backend -> do
               hydraScriptsTxId <- publishHydraScriptsAs backend Faucet
-              concurrently_
-                (initAndClose tmpDir tracer 0 hydraScriptsTxId backend)
-                (initAndClose tmpDir tracer 1 hydraScriptsTxId backend)
+              concurrentlyLabelled_
+                ("init-and-close-0", initAndClose tmpDir tracer 0 hydraScriptsTxId backend)
+                ("init-and-close-1", initAndClose tmpDir tracer 1 hydraScriptsTxId backend)
 
       it "alice inits a Head with incorrect keys preventing bob from observing InitTx" $ \tracer ->
         failAfter 60 $

hydra-node/test/Hydra/API/ServerSpec.hs

@@ -96,9 +96,9 @@ spec =
             semaphore <- newLabelledTVarIO "semaphore" 0
             withAsyncLabelled
               ( "concurrent-test-clients"
-              , concurrently_
-                  (withClient port "/" $ testClient queue semaphore)
-                  (withClient port "/" $ testClient queue semaphore)
+              , concurrentlyLabelled_
+                  ("concurrent-test-client-1", withClient port "/" $ testClient queue semaphore)
+                  ("concurrent-test-client-2", withClient port "/" $ testClient queue semaphore)
               )
               $ \_ -> do
                 waitForClients semaphore
@@ -131,9 +131,9 @@ spec =
             semaphore <- newLabelledTVarIO "semaphore" 0
             withAsyncLabelled
               ( "concurrent-test-clients"
-              , concurrently_
-                  (withClient port "/?history=yes" $ testClient queue1 semaphore)
-                  (withClient port "/?history=yes" $ testClient queue2 semaphore)
+              , concurrentlyLabelled_
+                  ("concurrent-test-client-queue1", withClient port "/?history=yes" $ testClient queue1 semaphore)
+                  ("concurrent-test-client-queue2", withClient port "/?history=yes" $ testClient queue2 semaphore)
               )
               $ \_ -> do
                 waitForClients semaphore

hydra-node/test/Hydra/Events/S3Spec.hs

@@ -53,9 +53,9 @@ spec = do
 
     it "allows concurrent usage" $ \bucketName -> do
       withS3EventStore bucketName $ \(source, sink) -> do
-        concurrently_
-          (putEvent sink 123)
-          (putEvent sink 456)
+        concurrentlyLabelled_
+          ("concurrent-put-event-123", putEvent sink 123)
+          ("concurrent-put-event-456", putEvent sink 456)
         getEvents source `shouldReturn` [123, 456 :: EventId]
 
     it "supports multiple instances" $ \bucketName ->

hydra-node/test/Hydra/Events/UDPSpec.hs

@@ -26,7 +26,7 @@ spec = do
     it "allows concurrent usage" $ do
       withFreePort $ \port -> do
         withUDPEventSink @EventId "0.0.0.0" (show port) $ \EventSink{putEvent} -> do
-          concurrently_ (putEvent 123) (putEvent 456)
+          concurrentlyLabelled_ ("put-event-123", putEvent 123) ("put-event-456", putEvent 456)
 
   it "supports multiple instances" $ do
     withFreePort $ \port -> do

hydra-prelude/src/Hydra/Prelude.hs

@@ -48,6 +48,8 @@ module Hydra.Prelude (
   newLabelledTQueueIO,
   newLabelledEmptyTMVarIO,
   labelMyThread,
+  concurrentlyLabelled,
+  concurrentlyLabelled_,
   raceLabelled,
   raceLabelled_,
   withAsyncLabelled,
@@ -363,3 +365,12 @@ raceLabelled_ = (void .) . raceLabelled
 
 withAsyncLabelled :: MonadAsync m => (String, m a) -> (Async m a -> m b) -> m b
 withAsyncLabelled (lbl, ma) = withAsync (labelMyThread lbl >> ma)
+
+concurrentlyLabelled :: (MonadThread m, MonadAsync m) => (String, m a) -> (String, m b) -> m (a, b)
+concurrentlyLabelled (lblA, mA) (lblB, mB) =
+  concurrently
+    (labelMyThread lblA >> mA)
+    (labelMyThread lblB >> mB)
+
+concurrentlyLabelled_ :: (MonadThread m, MonadAsync m) => (String, m a) -> (String, m b) -> m ()
+concurrentlyLabelled_ = (void .) . concurrentlyLabelled