potoki-core 2.2.5.3 → 2.2.5.4

raw patch · 3 files changed

+72/−45 lines, 3 filesdep −unagi-chandep ~stmPVP ok

version bump matches the API change (PVP)

Dependencies removed: unagi-chan

Dependency ranges changed: stm

API changes (from Hackage documentation)

Files

• library/Potoki/Core/Consume.hs +7/−6
• library/Potoki/Core/Transform/Concurrency.hs +64/−37
• potoki-core.cabal +1/−2

@@ -160,12 +160,13 @@ {-# INLINABLE count #-} count :: Consume input Int count =-  Consume $ \ (A.Fetch fetchIO) -> build fetchIO 0-  where-    build fetchIO !acc =-      fetchIO >>= \case-        Nothing -> pure acc-        Just _ -> build fetchIO (succ acc)+  Consume $ \ (A.Fetch fetchIO) -> let+    iterate !count = do+      fetchResult <- fetchIO+      case fetchResult of+        Just _ -> iterate (succ count)+        Nothing -> return count+    in iterate 0  {-# INLINABLE concat #-} concat :: (Semigroup monoid, Monoid monoid) => Consume monoid monoid

@@ -6,20 +6,32 @@ import Potoki.Core.Types import qualified Potoki.Core.Fetch as A import qualified Acquire.Acquire as M-import qualified Control.Concurrent.Chan.Unagi.Bounded as B   {-# INLINE bufferize #-} bufferize :: Int -> Transform element element bufferize size =-  Transform $ \ (A.Fetch fetch) -> M.Acquire $ do-    (inChan, outChan) <- B.newChan size-    forkIO $ fix $ \ doLoop ->-      fetch >>= \case-        Nothing -> B.writeChan inChan Nothing-        Just !element -> B.writeChan inChan (Just element) >> doLoop-    return $ (A.Fetch $ B.readChan outChan, return ())+  Transform $ \ (A.Fetch fetchIO) -> liftIO $ do+    buffer <- newTBQueueIO size+    activeVar <- newTVarIO True +    forkIO $ fix $ \ loop -> do+      fetchingResult <- fetchIO+      case fetchingResult of+        Just !element -> do+          atomically $ writeTBQueue buffer element+          loop+        Nothing -> atomically $ writeTVar activeVar False++    return $ Fetch $ let+      readBuffer = Just <$> readTBQueue buffer+      terminate = do+        active <- readTVar activeVar+        if active+          then empty+          else return Nothing+      in atomically (readBuffer <|> terminate)+ {-| Identity Transform, which ensures that the inputs are fetched synchronously. @@ -28,12 +40,12 @@ {-# INLINABLE sync #-} sync :: Transform a a sync =-  Transform $ \ (A.Fetch fetch) -> M.Acquire $ do+  Transform $ \ (A.Fetch fetchIO) -> liftIO $ do     activeVar <- newMVar True-    return $ (, return ()) $ A.Fetch $ do+    return $ A.Fetch $ do       active <- takeMVar activeVar       if active-        then fetch >>= \case+        then fetchIO >>= \ case           Just !element -> do             putMVar activeVar True             return (Just element)@@ -60,37 +72,52 @@ {-# INLINE concurrentlyUnsafe #-} concurrentlyUnsafe :: Int -> Transform input output -> Transform input output concurrentlyUnsafe workersAmount (Transform syncTransformIO) = -  Transform $ \ fetch -> M.Acquire $ do-    outChan <- newEmptyMVar+  Transform $ \ fetchIO -> liftIO $ do+    chan <- atomically newEmptyTMVar+    workersCounter <- atomically (newTVar workersAmount)+    fetchingAvailableVar <- atomically (newTVar True)+     replicateM_ workersAmount $ forkIO $ do-      let runAcquire (M.Acquire io) = io-      (A.Fetch fetchIO, _) <- runAcquire $ syncTransformIO fetch-      fix $ \ doLoop -> fetchIO >>= \case-        Nothing -> putMVar outChan Nothing-        Just !result -> putMVar outChan (Just result) >> doLoop-    activeWorkersAmountVar <- newMVar workersAmount-    return $ (, return ()) $ A.Fetch $ fix $ \ doLoop' -> do-      activeWorkersAmount <- takeMVar activeWorkersAmountVar-      if activeWorkersAmount <= 0-        then return Nothing-        else do-          fetchResult <- takeMVar outChan-          case fetchResult of-            Just result -> do-              putMVar activeWorkersAmountVar activeWorkersAmount-              return (Just result)-            Nothing -> do-              putMVar activeWorkersAmountVar (pred activeWorkersAmount)-              doLoop'+      (A.Fetch fetchIO, finalize) <- case syncTransformIO fetchIO of M.Acquire io -> io+      fix $ \ loop -> do+        fetchResult <- fetchIO+        case fetchResult of+          Just !result -> do+            atomically (putTMVar chan result)+            loop+          Nothing -> do+            atomically $ modifyTVar' workersCounter pred+      finalize +    return $ A.Fetch $ let+      readChan = Just <$> takeTMVar chan+      terminate = do+        workersActive <- readTVar workersCounter+        if workersActive > 0+          then empty+          else return Nothing+      in atomically (readChan <|> terminate)+ {-| A transform, which fetches the inputs asynchronously on the specified number of threads. -} async :: Int -> Transform input input async workersAmount = -  Transform $ \ (A.Fetch fetchIO) -> M.Acquire $ do-    chan <- newEmptyMVar -    replicateM_ workersAmount $ forkIO $ fix $ \ _ -> do+  Transform $ \ (A.Fetch fetchIO) -> liftIO $ do+    chan <- atomically newEmptyTMVar+    workersCounter <- atomically (newTVar workersAmount)++    replicateM_ workersAmount $ forkIO $ fix $ \ loop -> do       fetchResult <- fetchIO-      putMVar chan fetchResult-    return (A.finiteMVar chan, return ())+      case fetchResult of+        Just input -> atomically (putTMVar chan input) *> loop+        Nothing -> atomically (modifyTVar' workersCounter pred)++    return $ A.Fetch $ let+      readChan = Just <$> takeTMVar chan+      terminate = do+        workersActive <- readTVar workersCounter+        if workersActive > 0+          then empty+          else return Nothing+      in atomically (readChan <|> terminate)

@@ -1,7 +1,7 @@ name:   potoki-core version:-  2.2.5.3+  2.2.5.4 synopsis:   Low-level components of "potoki" description:@@ -72,7 +72,6 @@     stm >=2.4 && <3,     text >=1 && <2,     transformers >=0.5 && <0.6,-    unagi-chan >=0.4 && <0.5,     unordered-containers >=0.2 && <0.3,     vector >=0.12 && <0.13