@@ -74,16 +74,29 @@ module Yoga.Om.Strom
74 , zipWithStrom
74 , zipWithStrom
75 , interleave
75 , interleave
76 , intersperse
76 , intersperse
78 , mergeAll
79 , mergeAll
80 , mergeAllND
86 , mergeAllND
83 -- Parallel Processing
90 -- Parallel Processing
86 , foreachPar
93 , foreachPar
96+ , foreachParUnordered
87 -- Error Handling
100 -- Error Handling
88 , catchAll
101 , catchAll
89 -- , catchSome -- TODO: Fix type constraints
102 -- , catchSome -- TODO: Fix type constraints
@@ -115,7 +128,7 @@ import Data.Functor (map) as Functor
115import Data.List as List
128import Data.List as List
116import Data.Maybe (Maybe(..))
129import Data.Maybe (Maybe(..))
117import Data.Time.Duration (Milliseconds(..))
130import Data.Time.Duration (Milliseconds(..))
118−import Data.Traversable (traverse)
131+import Data.Traversable (traverse, traverse_)
119import Data.Tuple (Tuple(..))
132import Data.Tuple (Tuple(..))
120import Data.Variant (Variant)
133import Data.Variant (Variant)
121import Effect.Aff (Aff, Error, delay, forkAff, joinFiber, killFiber)
134import Effect.Aff (Aff, Error, delay, forkAff, joinFiber, killFiber)
@@ -123,7 +136,10 @@ import Effect.Aff.AVar as AVar
123import Effect.Aff.Class (liftAff)
136import Effect.Aff.Class (liftAff)
124import Effect.Exception as Exception
137import Effect.Exception as Exception
125import Effect.Now (now)
138import Effect.Now (now)
139+import Effect.Ref as Ref
126import Effect.Class (liftEffect)
140import Effect.Class (liftEffect)
141+import Data.CatQueue as CatQueue
142+import Data.CatQueue (CatQueue)
127import Data.DateTime.Instant (unInstant)
143import Data.DateTime.Instant (unInstant)
128import Yoga.Om (Om)
144import Yoga.Om (Om)
129import Yoga.Om as Om
145import Yoga.Om as Om
@@ -175,6 +191,29 @@ data QueueItem a
176derive instance functorQueueItem :: Functor QueueItem
192derive instance functorQueueItem :: Functor QueueItem
194+-- | Messages for mergeAllND buffered queue
196+ = MergeChunk (Array a)
199+derive instance functorMergeAllItem :: Functor MergeAllItem
201+-- | Merge termination strategy (like ZStream HaltStrategy)
208+derive instance eqHaltStrategy :: Eq HaltStrategy
210+-- | Messages for buffering queues
212+ = BufferChunk (Array a)
215+derive instance functorBufferItem :: Functor BufferItem
178--------------------------------------------------------------------------------
217--------------------------------------------------------------------------------
179-- Construction
218-- Construction
180--------------------------------------------------------------------------------
219--------------------------------------------------------------------------------
@@ -1097,41 +1136,187 @@ intersperse separator stream = intersperseHelper true stream
1097 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1136 else Array.cons separator (Array.intercalate [ separator ] (Array.singleton <$> chunk))
1098 pure $ Loop $ Tuple (Just withSeparators) (intersperseHelper false next)
1137 pure $ Loop $ Tuple (Just withSeparators) (intersperseHelper false next)
1139+--------------------------------------------------------------------------------
1141+--------------------------------------------------------------------------------
1143+-- | Buffer stream chunks with backpressure (preserves chunking)
1144+bufferChunks :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
1145+bufferChunks capacity stream
1146+ | capacity <= 0 = stream
1147+ | otherwise = mkStrom do
1150+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (BufferItem a))
1151+ sizeRef <- liftEffect $ Ref.new 0
1152+ mutex <- AVar.new unit
1153+ notEmpty <- AVar.empty
1154+ notFull <- AVar.new unit
1160+ size <- liftEffect $ Ref.read sizeRef
1161+ let wasEmpty = size == 0
1162+ buf <- liftEffect $ Ref.read bufferRef
1163+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1164+ liftEffect $ Ref.write (size + 1) sizeRef
1165+ AVar.put unit mutex
1166+ when wasEmpty $ AVar.put unit notEmpty
1167+ when (size + 1 < capacity) $ AVar.put unit notFull
1172+ buf <- liftEffect $ Ref.read bufferRef
1173+ size <- liftEffect $ Ref.read sizeRef
1174+ case CatQueue.uncons buf of
1176+ AVar.put unit mutex
1178+ Just (Tuple head tail) -> do
1179+ liftEffect $ Ref.write tail bufferRef
1180+ liftEffect $ Ref.write (size - 1) sizeRef
1181+ let hasMore = (size - 1) > 0
1182+ let wasFull = size == capacity
1183+ AVar.put unit mutex
1184+ when hasMore $ AVar.put unit notEmpty
1185+ when wasFull $ AVar.put unit notFull
1189+ stepResult <- Om.runReader ctx (runStrom s)
1190+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1192+ Done Nothing -> enqueue BufferDone
1193+ Done (Just chunk) -> do
1194+ enqueue (BufferChunk chunk)
1196+ Loop (Tuple maybeChunk next) -> do
1198+ Just chunk -> enqueue (BufferChunk chunk)
1199+ Nothing -> pure unit
1202+ fiber <- forkAff (producer stream)
1205+ consumer = mkStrom $ liftAff do
1209+ killFiber (Exception.error "done") fiber
1210+ pure $ Done Nothing
1211+ BufferChunk chunk ->
1212+ pure $ Loop $ Tuple (Just chunk) consumer
1214+ stepResult <- Om.runReader ctx (runStrom consumer)
1215+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1217+-- | Buffer stream elements with backpressure (destroys chunking)
1218+buffer :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a
1219+buffer capacity stream
1220+ | capacity <= 0 = stream
1221+ | otherwise = mkStrom do
1224+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (BufferItem a))
1225+ sizeRef <- liftEffect $ Ref.new 0
1226+ mutex <- AVar.new unit
1227+ notEmpty <- AVar.empty
1228+ notFull <- AVar.new unit
1234+ size <- liftEffect $ Ref.read sizeRef
1235+ let wasEmpty = size == 0
1236+ buf <- liftEffect $ Ref.read bufferRef
1237+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1238+ liftEffect $ Ref.write (size + 1) sizeRef
1239+ AVar.put unit mutex
1240+ when wasEmpty $ AVar.put unit notEmpty
1241+ when (size + 1 < capacity) $ AVar.put unit notFull
1246+ buf <- liftEffect $ Ref.read bufferRef
1247+ size <- liftEffect $ Ref.read sizeRef
1248+ case CatQueue.uncons buf of
1250+ AVar.put unit mutex
1252+ Just (Tuple head tail) -> do
1253+ liftEffect $ Ref.write tail bufferRef
1254+ liftEffect $ Ref.write (size - 1) sizeRef
1255+ let hasMore = (size - 1) > 0
1256+ let wasFull = size == capacity
1257+ AVar.put unit mutex
1258+ when hasMore $ AVar.put unit notEmpty
1259+ when wasFull $ AVar.put unit notFull
1263+ stepResult <- Om.runReader ctx (runStrom s)
1264+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1266+ Done Nothing -> enqueue BufferDone
1267+ Done (Just chunk) -> do
1268+ traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1270+ Loop (Tuple maybeChunk next) -> do
1272+ Just chunk -> traverse_ (\a -> enqueue (BufferChunk [ a ])) chunk
1273+ Nothing -> pure unit
1276+ fiber <- forkAff (producer stream)
1279+ consumer = mkStrom $ liftAff do
1283+ killFiber (Exception.error "done") fiber
1284+ pure $ Done Nothing
1285+ BufferChunk chunk ->
1286+ pure $ Loop $ Tuple (Just chunk) consumer
1288+ stepResult <- Om.runReader ctx (runStrom consumer)
1289+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1100--------------------------------------------------------------------------------
1291--------------------------------------------------------------------------------
1101-- Concurrency
1292-- Concurrency
1102--------------------------------------------------------------------------------
1293--------------------------------------------------------------------------------
1104−-- | Merge two streams concurrently
1295+-- | Merge two streams concurrently (nondeterministic like ZStream merge)
1105-- | Elements are emitted as soon as they're available from either stream
1296-- | Elements are emitted as soon as they're available from either stream
1106merge :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1297merge :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1107−merge s1 s2 = mkStrom do
1298+merge = mergeNDWith 1024
1108− -- Pull from both streams concurrently
1299+
1109− step1 <- runStrom s1
1300+-- | Merge with halt strategy and mapping functions (ZStream-like)
1110− step2 <- runStrom s2
1301+mergeWith
1111− case step1, step2 of
1302+ :: forall ctx err a b c
1112− Done Nothing, Done Nothing -> pure $ Done Nothing
1303+ . HaltStrategy
1113− Done Nothing, Done (Just chunk2) -> pure $ Done $ Just chunk2
1304+ -> (a -> c)
1114− Done (Just chunk1), Done Nothing -> pure $ Done $ Just chunk1
1305+ -> (b -> c)
1115− Done (Just chunk1), Done (Just chunk2) -> pure $ Done $ Just (chunk1 <> chunk2)
1306+ -> Strom ctx err a
1116− Done Nothing, Loop (Tuple maybeChunk2 next2) ->
1307+ -> Strom ctx err b
1117− pure $ Loop $ Tuple maybeChunk2 next2
1308+ -> Strom ctx err c
1118− Done (Just chunk1), Loop (Tuple maybeChunk2 next2) ->
1309+mergeWith strategy f g s1 s2 =
1119− case maybeChunk2 of
1310+ mergeNDWithStrategy 1024 strategy (mapStrom f s1) (mapStrom g s2)
1120− Nothing -> pure $ Loop $ Tuple (Just chunk1) next2
1311+
1121− Just chunk2 -> pure $ Loop $ Tuple (Just $ chunk1 <> chunk2) next2
1312+mergeHaltEither :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1122− Loop (Tuple maybeChunk1 next1), Done Nothing ->
1313+mergeHaltEither = mergeNDWithStrategy 1024 HaltEither
1123− pure $ Loop $ Tuple maybeChunk1 next1
1314+
1124− Loop (Tuple maybeChunk1 next1), Done (Just chunk2) ->
1315+mergeHaltLeft :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1125− case maybeChunk1 of
1316+mergeHaltLeft = mergeNDWithStrategy 1024 HaltLeft
1126− Nothing -> pure $ Loop $ Tuple (Just chunk2) next1
1317+
1127− Just chunk1 -> pure $ Loop $ Tuple (Just $ chunk1 <> chunk2) next1
1318+mergeHaltRight :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1128− Loop (Tuple maybeChunk1 next1), Loop (Tuple maybeChunk2 next2) ->
1319+mergeHaltRight = mergeNDWithStrategy 1024 HaltRight
1129− -- Both streams continue, merge chunks and continue with both
1130− case maybeChunk1, maybeChunk2 of
1131− Nothing, Nothing -> pure $ Loop $ Tuple Nothing (merge next1 next2)
1132− Just c1, Nothing -> pure $ Loop $ Tuple (Just c1) (merge next1 next2)
1133− Nothing, Just c2 -> pure $ Loop $ Tuple (Just c2) (merge next1 next2)
1134− Just c1, Just c2 -> pure $ Loop $ Tuple (Just $ c1 <> c2) (merge next1 next2)
1136-- | Merge multiple streams concurrently
1321-- | Merge multiple streams concurrently
1137mergeAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1322mergeAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
@@ -1140,27 +1325,83 @@ mergeAll streams = Array.foldl merge empty streams
1140-- | Race two streams - returns elements from whichever produces first
1325-- | Race two streams - returns elements from whichever produces first
1141-- | Uses Om.race for true concurrent racing
1326-- | Uses Om.race for true concurrent racing
1142race :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1327race :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1143−race s1 s2 = mkStrom do
1328+race = mergeHaltEither
1144− Om.race [ runStrom s1, runStrom s2 ]
1146-- | Race multiple streams - returns elements from whichever produces first
1330-- | Race multiple streams - returns elements from whichever produces first
1147raceAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1331raceAll :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1148−raceAll streams = mkStrom do
1332+raceAll streams =
1149− Om.race (Functor.map runStrom streams)
1333+ case Array.uncons streams of
1335+ Just { head, tail } -> Array.foldl mergeHaltEither head tail
1151-- | Non-deterministic merge: Merge two streams concurrently, emitting elements as soon as they're available
1337-- | Non-deterministic merge: Merge two streams concurrently, emitting elements as soon as they're available
1152-- | Unlike `merge`, this runs both streams in parallel and elements may arrive in any order
1338-- | Unlike `merge`, this runs both streams in parallel and elements may arrive in any order
1153-- | The stream that produces results faster will have its elements appear first
1339-- | The stream that produces results faster will have its elements appear first
1154-- | Optimized: Spawns producer fibers once and uses a queue for communication
1340-- | Optimized: Spawns producer fibers once and uses a queue for communication
1155mergeND :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1341mergeND :: forall ctx err a. Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1342+mergeND = mergeNDWith 1024
1344+-- | Non-deterministic merge with configurable buffer capacity (chunk queue)
1345+mergeNDWith :: forall ctx err a. Int -> Strom ctx err a -> Strom ctx err a -> Strom ctx err a
1346+mergeNDWith bufferCapacity = mergeNDWithStrategy bufferCapacity HaltBoth
1349+ :: forall ctx err a
1156−mergeND stream1 stream2 = mkStrom do
1355+mergeNDWithStrategy bufferCapacity strategy stream1 stream2 = mkStrom do
1157 -- Capture context
1356 -- Capture context
1158 ctx <- Om.ask
1357 ctx <- Om.ask
1160 liftAff do
1359 liftAff do
1161− -- Create communication queue
1360+ -- Create buffered queue with backpressure
1162− queue <- AVar.empty
1361+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (QueueItem a))
1163−
1362+ sizeRef <- liftEffect $ Ref.new 0
1363+ mutex <- AVar.new unit
1364+ notEmpty <- AVar.empty
1365+ notFull <- AVar.new unit
1368+ shouldStop doneLeft doneRight =
1370+ HaltBoth -> doneLeft && doneRight
1371+ HaltEither -> doneLeft || doneRight
1372+ HaltLeft -> doneLeft
1373+ HaltRight -> doneRight
1377+ size <- liftEffect $ Ref.read sizeRef
1378+ let wasEmpty = size == 0
1379+ buf <- liftEffect $ Ref.read bufferRef
1380+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1381+ liftEffect $ Ref.write (size + 1) sizeRef
1382+ AVar.put unit mutex
1383+ when wasEmpty $ AVar.put unit notEmpty
1384+ when (size + 1 < bufferCapacity) $ AVar.put unit notFull
1389+ buf <- liftEffect $ Ref.read bufferRef
1390+ size <- liftEffect $ Ref.read sizeRef
1391+ case CatQueue.uncons buf of
1393+ AVar.put unit mutex
1395+ Just (Tuple head tail) -> do
1396+ liftEffect $ Ref.write tail bufferRef
1397+ liftEffect $ Ref.write (size - 1) sizeRef
1398+ let hasMore = (size - 1) > 0
1399+ let wasFull = size == bufferCapacity
1400+ AVar.put unit mutex
1401+ when hasMore $ AVar.put unit notEmpty
1402+ when wasFull $ AVar.put unit notFull
1164 -- Spawn producer fiber for stream1
1405 -- Spawn producer fiber for stream1
1165 fiber1 <- forkAff do
1406 fiber1 <- forkAff do
@@ -1168,18 +1409,18 @@ mergeND stream1 stream2 = mkStrom do
1168 stepResult <- Om.runReader ctx (runStrom s)
1409 stepResult <- Om.runReader ctx (runStrom s)
1169 step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1410 step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1170 case step of
1411 case step of
1171− Done Nothing ->
1412+ Done Nothing ->
1172− AVar.put (Left StreamId1) queue -- Signal stream1 done
1413+ enqueue (StreamDone StreamId1)
1173 Done (Just chunk) -> do
1414 Done (Just chunk) -> do
1174− AVar.put (Right chunk) queue -- Emit final chunk
1415+ enqueue (DataChunk chunk StreamId1)
1175− AVar.put (Left StreamId1) queue -- Signal stream1 done
1416+ enqueue (StreamDone StreamId1)
1176 Loop (Tuple maybeChunk next) -> do
1417 Loop (Tuple maybeChunk next) -> do
1177 case maybeChunk of
1418 case maybeChunk of
1178− Just chunk -> AVar.put (Right chunk) queue
1419+ Just chunk -> enqueue (DataChunk chunk StreamId1)
1179 Nothing -> pure unit
1420 Nothing -> pure unit
1180 producer next
1421 producer next
1181 producer stream1
1422 producer stream1
1183 -- Spawn producer fiber for stream2
1424 -- Spawn producer fiber for stream2
1184 fiber2 <- forkAff do
1425 fiber2 <- forkAff do
@@ -1187,44 +1428,126 @@ mergeND stream1 stream2 = mkStrom do
1187 stepResult <- Om.runReader ctx (runStrom s)
1428 stepResult <- Om.runReader ctx (runStrom s)
1188 step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1429 step <- either (\e -> throwError (Exception.error "Stream error")) pure stepResult
1189 case step of
1430 case step of
1190− Done Nothing ->
1431+ Done Nothing ->
1191− AVar.put (Left StreamId2) queue -- Signal stream2 done
1432+ enqueue (StreamDone StreamId2)
1192 Done (Just chunk) -> do
1433 Done (Just chunk) -> do
1193− AVar.put (Right chunk) queue -- Emit final chunk
1434+ enqueue (DataChunk chunk StreamId2)
1194− AVar.put (Left StreamId2) queue -- Signal stream2 done
1435+ enqueue (StreamDone StreamId2)
1195 Loop (Tuple maybeChunk next) -> do
1436 Loop (Tuple maybeChunk next) -> do
1196 case maybeChunk of
1437 case maybeChunk of
1197− Just chunk -> AVar.put (Right chunk) queue
1438+ Just chunk -> enqueue (DataChunk chunk StreamId2)
1198 Nothing -> pure unit
1439 Nothing -> pure unit
1199 producer next
1440 producer next
1200 producer stream2
1441 producer stream2
1202 -- Consumer: pull from queue
1443 -- Consumer: pull from queue
1204− consumer doneCount = mkStrom $ liftAff do
1445+ consumer doneLeft doneRight = mkStrom $ liftAff do
1205− if doneCount >= 2 then do
1446+ if shouldStop doneLeft doneRight then do
1206 -- Both streams done, cleanup
1447 -- Both streams done, cleanup
1207 killFiber (Exception.error "done") fiber1
1448 killFiber (Exception.error "done") fiber1
1208 killFiber (Exception.error "done") fiber2
1449 killFiber (Exception.error "done") fiber2
1209 pure $ Done Nothing
1450 pure $ Done Nothing
1211− msg <- AVar.take queue
1452+ msg <- dequeue
1212 case msg of
1453 case msg of
1213− Left _streamId -> do
1454+ StreamDone streamId -> do
1214 -- One stream finished, recurse
1455 -- One stream finished, recurse
1457+ doneLeft' = doneLeft || streamId == StreamId1
1458+ doneRight' = doneRight || streamId == StreamId2
1215− stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
1459+ stepResult <- Om.runReader ctx (runStrom $ consumer doneLeft' doneRight')
1216 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1460 either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1217− Right chunk ->
1461+ DataChunk chunk _streamId ->
1218 -- Got a chunk
1462 -- Got a chunk
1219− pure $ Loop $ Tuple (Just chunk) (consumer doneCount)
1463+ pure $ Loop $ Tuple (Just chunk) (consumer doneLeft doneRight)
1221− stepResult <- Om.runReader ctx (runStrom $ consumer 0)
1465+ stepResult <- Om.runReader ctx (runStrom $ consumer false false)
1222 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1466 either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1224-- | Non-deterministic merge of multiple streams
1468-- | Non-deterministic merge of multiple streams
1225-- | All streams run concurrently and emit elements as soon as they're available
1469-- | All streams run concurrently and emit elements as soon as they're available
1226mergeAllND :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1470mergeAllND :: forall ctx err a. Array (Strom ctx err a) -> Strom ctx err a
1227−mergeAllND streams = Array.foldl mergeND empty streams
1471+mergeAllND = mergeAllNDWith 1024
1473+-- | Non-deterministic merge of multiple streams with configurable buffer capacity
1474+mergeAllNDWith :: forall ctx err a. Int -> Array (Strom ctx err a) -> Strom ctx err a
1475+mergeAllNDWith bufferCapacity streams
1476+ | Array.null streams = empty
1477+ | otherwise = mkStrom do
1480+ bufferRef <- liftEffect $ Ref.new (CatQueue.empty :: CatQueue (MergeAllItem a))
1481+ sizeRef <- liftEffect $ Ref.new 0
1482+ mutex <- AVar.new unit
1483+ notEmpty <- AVar.empty
1484+ notFull <- AVar.new unit
1490+ size <- liftEffect $ Ref.read sizeRef
1491+ let wasEmpty = size == 0
1492+ buf <- liftEffect $ Ref.read bufferRef
1493+ liftEffect $ Ref.write (CatQueue.snoc buf item) bufferRef
1494+ liftEffect $ Ref.write (size + 1) sizeRef
1495+ AVar.put unit mutex
1496+ when wasEmpty $ AVar.put unit notEmpty
1497+ when (size + 1 < bufferCapacity) $ AVar.put unit notFull
1502+ buf <- liftEffect $ Ref.read bufferRef
1503+ size <- liftEffect $ Ref.read sizeRef
1504+ case CatQueue.uncons buf of
1506+ AVar.put unit mutex
1508+ Just (Tuple head tail) -> do
1509+ liftEffect $ Ref.write tail bufferRef
1510+ liftEffect $ Ref.write (size - 1) sizeRef
1511+ let hasMore = (size - 1) > 0
1512+ let wasFull = size == bufferCapacity
1513+ AVar.put unit mutex
1514+ when hasMore $ AVar.put unit notEmpty
1515+ when wasFull $ AVar.put unit notFull
1519+ stepResult <- Om.runReader ctx (runStrom s)
1520+ step <- either (\_ -> throwError (Exception.error "Stream error")) pure stepResult
1522+ Done Nothing -> enqueue MergeDone
1523+ Done (Just chunk) -> do
1524+ enqueue (MergeChunk chunk)
1526+ Loop (Tuple maybeChunk next) -> do
1528+ Just chunk -> enqueue (MergeChunk chunk)
1529+ Nothing -> pure unit
1532+ fibers <- traverse (\s -> forkAff (producer s)) streams
1535+ totalStreams = Array.length streams
1536+ consumer doneCount = mkStrom $ liftAff do
1537+ if doneCount >= totalStreams then do
1538+ traverse_ (\fiber -> killFiber (Exception.error "done") fiber) fibers
1539+ pure $ Done Nothing
1544+ stepResult <- Om.runReader ctx (runStrom $ consumer (doneCount + 1))
1545+ either (\_ -> throwError (Exception.error "Consumer error")) pure stepResult
1546+ MergeChunk chunk ->
1547+ pure $ Loop $ Tuple (Just chunk) (consumer doneCount)
1549+ stepResult <- Om.runReader ctx (runStrom $ consumer 0)
1550+ either (\_ -> throwError (Exception.error "Consumer init error")) pure stepResult
1229--------------------------------------------------------------------------------
1552--------------------------------------------------------------------------------
1230-- Parallel Processing
1553-- Parallel Processing
@@ -1264,14 +1587,26 @@ mapPar concurrency f stream
1265 if Array.null group then [] else Array.cons group (chunksOf n rest)
1588 if Array.null group then [] else Array.cons group (chunksOf n rest)
1590+-- | Unordered parallel map (currently same as mapPar)
1591+mapParUnordered :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1592+mapParUnordered = mapPar
1267-- | Alias for mapPar with explicit "M" naming
1594-- | Alias for mapPar with explicit "M" naming
1268mapMPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1595mapMPar :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1269mapMPar = mapPar
1596mapMPar = mapPar
1598+-- | Alias for unordered mapPar with explicit "M" naming
1599+mapMParUnordered :: forall ctx err a b. Int -> (a -> Om ctx err b) -> Strom ctx err a -> Strom ctx err b
1600+mapMParUnordered = mapParUnordered
1271-- | For each element, execute an effect in parallel with bounded concurrency
1602-- | For each element, execute an effect in parallel with bounded concurrency
1272foreachPar :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
1603foreachPar :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
1273foreachPar concurrency f stream = runDrain (mapPar concurrency f stream)
1604foreachPar concurrency f stream = runDrain (mapPar concurrency f stream)
1606+-- | Unordered parallel foreach
1607+foreachParUnordered :: forall ctx err a. Int -> (a -> Om ctx err Unit) -> Strom ctx err a -> Om ctx err Unit
1608+foreachParUnordered concurrency f stream = runDrain (mapParUnordered concurrency f stream)
1275--------------------------------------------------------------------------------
1610--------------------------------------------------------------------------------
1276-- Error Handling
1611-- Error Handling
1277--------------------------------------------------------------------------------
1612--------------------------------------------------------------------------------