streaming-concurrency (empty) → 0.1.0.0
raw patch · 8 files changed
+513/−0 lines, 8 filesdep +QuickCheckdep +basedep +bytestringsetup-changed
Dependencies added: QuickCheck, base, bytestring, exceptions, hspec, lifted-async, monad-control, quickcheck-instances, stm, streaming, streaming-bytestring, streaming-concurrency, streaming-with, transformers-base
Files
- ChangeLog.md +5/−0
- LICENSE +20/−0
- README.md +23/−0
- Setup.hs +2/−0
- src/Streaming/Concurrent.hs +253/−0
- src/Streaming/Concurrent/Lifted.hs +106/−0
- streaming-concurrency.cabal +53/−0
- test/merging.hs +51/−0
+ ChangeLog.md view
@@ -0,0 +1,5 @@+# Revision history for streaming-concurrency++## 0.1.0.0 -- 2017-07-04++* First version.
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2017 Ivan Lazar Miljenovic++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ README.md view
@@ -0,0 +1,23 @@+streaming-concurrency+==============++[](https://hackage.haskell.org/package/streaming-concurrency) [](https://travis-ci.org/ivan-m/streaming-concurrency)++> Concurrency for the [streaming] ecosystem++The primary purpose for this library is to be able to merge multiple+`Stream`s together. However, it is possible to build higher+abstractions on top of this to be able to also feed multiple streams.++[streaming]: http://hackage.haskell.org/package/streaming++Thanks and recognition+----------------------++The code here is heavily based upon -- and borrows the underlying+`Buffer` code from -- Gabriel Gonzalez's [pipes-concurrency]. It+differs from it primarily in being more bracket-oriented rather than+providing a `spawn` primitive, thus not requiring explicit garbage+collection.++[pipes-concurrency]: http://hackage.haskell.org/package/pipes-concurrency
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ src/Streaming/Concurrent.hs view
@@ -0,0 +1,253 @@+{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses #-}++{- |+ Module : Streaming.Concurrent+ Description : Concurrency support for the streaming ecosystem+ Copyright : Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++ Consider a physical desk for someone that has to deal with+ correspondence.++ A typical system is to have two baskets\/trays: one for incoming+ papers that still needs to be processed, and another for outgoing+ papers that have already been processed.++ We use this metaphor for dealing with 'Buffer's: data is fed into+ one using the 'InBasket' (until the buffer indicates that it has+ had enough) and taken out from the 'OutBasket'.++ -}+module Streaming.Concurrent+ ( -- * Buffers+ Buffer+ , unbounded+ , bounded+ , latest+ , newest+ -- * Using a buffer+ , withBuffer+ , InBasket(..)+ , OutBasket(..)+ -- * Stream support+ , writeStreamBasket+ , readStreamBasket+ , mergeStreams+ -- * ByteString support+ , writeByteStringBasket+ , readByteStringBasket+ , mergeByteStrings+ ) where++import Data.ByteString.Streaming (ByteString, reread, unconsChunk)+import Streaming (Of, Stream)+import qualified Streaming.Prelude as S++import Control.Applicative ((<|>))+import Control.Concurrent.Async.Lifted (concurrently,+ forConcurrently_)+import qualified Control.Concurrent.STM as STM+import Control.Monad (when)+import Control.Monad.Base (MonadBase, liftBase)+import Control.Monad.Catch (MonadMask, bracket, bracket_)+import Control.Monad.Trans.Control (MonadBaseControl)+import qualified Data.ByteString as B+import Data.Foldable (forM_)++--------------------------------------------------------------------------------++-- | Concurrently merge multiple streams together.+--+-- The resulting order is unspecified.+--+-- Note that the monad of the resultant Stream can be different from+-- the final result.+mergeStreams :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t)+ => Buffer a -> t (Stream (Of a) m v)+ -> (Stream (Of a) n () -> m r) -> m r+mergeStreams buff strs f = withBuffer buff+ (forConcurrently_ strs . flip writeStreamBasket)+ (`readStreamBasket` f)++-- | A streaming 'ByteString' variant of 'mergeStreams'.+mergeByteStrings :: (MonadMask m, MonadBaseControl IO m, MonadBase IO n, Foldable t)+ => Buffer B.ByteString -> t (ByteString m v)+ -> (ByteString n () -> m r) -> m r+mergeByteStrings buff bss f = withBuffer buff+ (forConcurrently_ bss . flip writeByteStringBasket)+ (`readByteStringBasket` f)++-- | Write a single stream to a buffer.+--+-- Type written to make it easier if this is the only stream being+-- written to the buffer.+writeStreamBasket :: (MonadBase IO m) => Stream (Of a) m r -> InBasket a -> m ()+writeStreamBasket stream (InBasket send) = go stream+ where+ go str = do eNxt <- S.next str -- uncons requires r ~ ()+ forM_ eNxt $ \(a, str') -> do+ continue <- liftBase (STM.atomically (send a))+ when continue (go str')++-- | A streaming 'ByteString' variant of 'writeStreamBasket'.+writeByteStringBasket :: (MonadBase IO m) => ByteString m r -> InBasket B.ByteString -> m ()+writeByteStringBasket bstring (InBasket send) = go bstring+ where+ go bs = do chNxt <- unconsChunk bs+ forM_ chNxt $ \(chnk, bs') -> do+ continue <- liftBase (STM.atomically (send chnk))+ when continue (go bs')++-- | Read the output of a buffer into a stream.+readStreamBasket :: (MonadBase IO m) => OutBasket a+ -> (Stream (Of a) m () -> r)+ -> r+readStreamBasket (OutBasket receive) f = f (S.untilRight getNext)+ where+ getNext = maybe (Right ()) Left <$> liftBase (STM.atomically receive)++-- | A streaming 'ByteString' variant of 'readStreamBasket'.+readByteStringBasket :: (MonadBase IO m) => OutBasket B.ByteString+ -> (ByteString m () -> r)+ -> r+readByteStringBasket (OutBasket receive) f =+ f (reread (liftBase . STM.atomically) receive)++--------------------------------------------------------------------------------+-- This entire section is almost completely taken from+-- pipes-concurrent by Gabriel Gonzalez:+-- https://github.com/Gabriel439/Haskell-Pipes-Concurrency-Library++-- | 'Buffer' specifies how to buffer messages between our 'InBasket'+-- and our 'OutBasket'.+data Buffer a+ = Unbounded+ | Bounded Int+ | Single+ | Latest a+ | Newest Int+ | New++-- | Store an unbounded number of messages in a FIFO queue.+unbounded :: Buffer a+unbounded = Unbounded++-- | Store a bounded number of messages, specified by the 'Int'+-- argument.+--+-- A buffer size @<= 0@ will result in a permanently empty buffer,+-- which could result in a system that hang.+bounded :: Int -> Buffer a+bounded 1 = Single+bounded n = Bounded n++-- | Only store the \"latest\" message, beginning with an initial+-- value.+--+-- This buffer is never empty nor full; as such, it is up to the+-- caller to ensure they only take as many values as they need+-- (e.g. using @'S.print' . 'readStreamBasket'@ as the final+-- parameter to 'withBuffer' will -- after all other values are+-- processed -- keep printing the last value over and over again).+latest :: a -> Buffer a+latest = Latest++-- | Like 'bounded', but 'sendMsg' never fails (the buffer is never+-- full). Instead, old elements are discard to make room for new+-- elements.+--+-- As with 'bounded', providing a size @<= 0@ will result in no+-- values being provided to the buffer, thus no values being read+-- and hence the system will most likely hang.+newest :: Int -> Buffer a+newest 1 = New+newest n = Newest n++-- | An exhaustible source of values.+--+-- 'receiveMsg' returns 'Nothing' if the source is exhausted.+newtype OutBasket a = OutBasket { receiveMsg :: STM.STM (Maybe a) }++-- | An exhaustible sink of values.+--+-- 'sendMsg' returns 'False' if the sink is exhausted.+newtype InBasket a = InBasket { sendMsg :: a -> STM.STM Bool }++-- | Use a buffer to asynchronously communicate.+--+-- Two functions are taken as parameters:+--+-- * How to provide input to the buffer (the result of this is+-- discarded)+--+-- * How to take values from the buffer+--+-- As soon as one function indicates that it is complete then the+-- other is terminated. This is safe: trying to write data to a+-- closed buffer will not achieve anything.+--+-- However, reading a buffer that has not indicated that it is+-- closed (e.g. waiting on an action to complete to be able to+-- provide the next value) but contains no values will block.+withBuffer :: (MonadMask m, MonadBaseControl IO m)+ => Buffer a -> (InBasket a -> m i)+ -> (OutBasket a -> m r) -> m r+withBuffer buffer sendIn readOut =+ bracket+ (liftBase openBasket)+ (\(_, _, _, seal) -> liftBase (STM.atomically seal)) $+ \(writeB, readB, sealed, seal) ->+ snd <$> concurrently (withIn writeB sealed seal)+ (withOut readB sealed seal)+ where+ openBasket = do+ (writeB, readB) <- case buffer of+ Bounded n -> do+ q <- STM.newTBQueueIO n+ return (STM.writeTBQueue q, STM.readTBQueue q)+ Unbounded -> do+ q <- STM.newTQueueIO+ return (STM.writeTQueue q, STM.readTQueue q)+ Single -> do+ m <- STM.newEmptyTMVarIO+ return (STM.putTMVar m, STM.takeTMVar m)+ Latest a -> do+ t <- STM.newTVarIO a+ return (STM.writeTVar t, STM.readTVar t)+ New -> do+ m <- STM.newEmptyTMVarIO+ return (\x -> STM.tryTakeTMVar m *> STM.putTMVar m x, STM.takeTMVar m)+ Newest n -> do+ q <- STM.newTBQueueIO n+ let writeB x = STM.writeTBQueue q x <|> (STM.tryReadTBQueue q *> writeB x)+ return (writeB, STM.readTBQueue q)++ -- We use this TVar as the communication mechanism between+ -- inputs and outputs as to whether either sub-continuation has+ -- finished.+ sealed <- STM.newTVarIO False+ let seal = STM.writeTVar sealed True++ return (writeB, readB, sealed, seal)++ withIn writeB sealed seal =+ bracket_ (return ())+ (liftBase (STM.atomically seal))+ (sendIn (InBasket sendOrEnd))+ where+ sendOrEnd a = do+ canWrite <- not <$> STM.readTVar sealed+ when canWrite (writeB a)+ return canWrite++ withOut readB sealed seal =+ bracket_ (return ())+ (liftBase (STM.atomically seal))+ (readOut (OutBasket readOrEnd))+ where+ readOrEnd = (Just <$> readB) <|> (do+ b <- STM.readTVar sealed+ STM.check b+ return Nothing )+{-# INLINABLE withBuffer #-}
+ src/Streaming/Concurrent/Lifted.hs view
@@ -0,0 +1,106 @@+{-# LANGUAGE FlexibleContexts, MultiParamTypeClasses #-}++{- |+ Module : Streaming.Concurrent.Lifted+ Description : Lifted variants of functions in "Streaming.Concurrent"+ Copyright : Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++ This module defines variants of those in "Streaming.Concurrent" for+ use with the 'Withable' class, found in the @streaming-with@+ package.++ -}+module Streaming.Concurrent.Lifted+ ( -- * Buffers+ Buffer+ , unbounded+ , bounded+ , latest+ , newest+ -- * Using a buffer+ , withBuffer+ , InBasket(..)+ , OutBasket(..)+ -- * Stream support+ , writeStreamBasket+ , readStreamBasket+ , mergeStreams+ -- * ByteString support+ , writeByteStringBasket+ , readByteStringBasket+ , mergeByteStrings+ ) where++import Data.ByteString.Streaming (ByteString)+import Streaming (Of, Stream)+import Streaming.Concurrent (Buffer, InBasket(..), OutBasket(..),+ bounded, latest, newest, unbounded)+import qualified Streaming.Concurrent as SC+import Streaming.With.Lifted (Withable(..))++import Control.Monad.Base (MonadBase)+import Control.Monad.Trans.Control (MonadBaseControl)++import qualified Data.ByteString as B++--------------------------------------------------------------------------------++-- | Concurrently merge multiple streams together.+--+-- The resulting order is unspecified.+mergeStreams :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO m, Foldable t)+ => Buffer a -> t (Stream (Of a) (WithMonad w) v)+ -> w (Stream (Of a) m ())+mergeStreams buff strs = liftWith (SC.mergeStreams buff strs)++-- | A streaming 'ByteString' variant of 'mergeStreams'.+mergeByteStrings :: (Withable w, MonadBaseControl IO (WithMonad w), MonadBase IO n, Foldable t)+ => Buffer B.ByteString -> t (ByteString (WithMonad w) v)+ -> w (ByteString n ())+mergeByteStrings buff bss = liftWith (SC.mergeByteStrings buff bss)++-- | Write a single stream to a buffer.+--+-- Type written to make it easier if this is the only stream being+-- written to the buffer.+writeStreamBasket :: (Withable w, MonadBase IO (WithMonad w))+ => Stream (Of a) (WithMonad w) r -> InBasket a -> w ()+writeStreamBasket stream ib = liftAction (SC.writeStreamBasket stream ib)++-- | A streaming 'ByteString' variant of 'writeStreamBasket'.+writeByteStringBasket :: (Withable w, MonadBase IO (WithMonad w))+ => ByteString (WithMonad w) r -> InBasket B.ByteString -> w ()+writeByteStringBasket bs ib = liftAction (SC.writeByteStringBasket bs ib)++-- | Read the output of a buffer into a stream.+--+-- Note that there is no requirement that @m ~ WithMonad w@.+readStreamBasket :: (Withable w, MonadBase IO m) => OutBasket a -> w (Stream (Of a) m ())+readStreamBasket ob = liftWith (SC.readStreamBasket ob)++-- | A streaming 'ByteString' variant of 'readStreamBasket'.+readByteStringBasket :: (Withable w, MonadBase IO m)+ => OutBasket B.ByteString -> w (ByteString m ())+readByteStringBasket ob = liftWith (SC.readByteStringBasket ob)++-- | Use a buffer to asynchronously communicate.+--+-- Two functions are taken as parameters:+--+-- * How to provide input to the buffer (the result of this is+-- discarded)+--+-- * How to take values from the buffer+--+-- As soon as one function indicates that it is complete then the+-- other is terminated. This is safe: trying to write data to a+-- closed buffer will not achieve anything.+--+-- However, reading a buffer that has not indicated that it is+-- closed (e.g. waiting on an action to complete to be able to+-- provide the next value) but contains no values will block.+withBuffer :: (Withable w, MonadBaseControl IO (WithMonad w))+ => Buffer a -> (InBasket a -> WithMonad w i) -> w (OutBasket a)+withBuffer buffer sendIn = liftWith (SC.withBuffer buffer sendIn)
+ streaming-concurrency.cabal view
@@ -0,0 +1,53 @@+name: streaming-concurrency+version: 0.1.0.0+synopsis: Concurrency support for the streaming ecosystem+description:+ The primary purpose for this library is to be able to merge multiple+ @Stream@s together. However, it is possible to build higher+ abstractions on top of this to be able to also feed multiple+ streams.+license: MIT+license-file: LICENSE+author: Ivan Lazar Miljenovic+maintainer: Ivan.Miljenovic@gmail.com+copyright: Ivan Lazar Miljenovic+category: Data+build-type: Simple+extra-source-files: ChangeLog.md, README.md+cabal-version: >=1.10+tested-with: GHC == 7.10.2, GHC == 8.0.2, GHC == 8.1.*++source-repository head+ type: git+ location: https://github.com/ivan-m/streaming-concurrency.git++library+ exposed-modules: Streaming.Concurrent+ , Streaming.Concurrent.Lifted+ build-depends: base ==4.*+ , bytestring+ , exceptions >= 0.6 && < 0.9+ , lifted-async >= 0.9.1 && < 0.10+ , monad-control == 1.*+ , stm >= 2.4 && < 3+ , streaming >= 0.1.4.0 && < 0.2+ , streaming-bytestring >= 0.1.4.5 && < 0.2+ , streaming-with == 0.1.*+ , transformers-base+ hs-source-dirs: src+ default-language: Haskell2010++test-suite merging+ type: exitcode-stdio-1.0+ main-is: merging.hs+ build-depends: streaming-concurrency+ , base+ , bytestring+ , hspec == 2.4.*+ , QuickCheck == 2.*+ , quickcheck-instances+ , streaming+ , streaming-bytestring+ hs-source-dirs: test+ default-language: Haskell2010+ ghc-options: -Wall
+ test/merging.hs view
@@ -0,0 +1,51 @@+{- |+ Module : Main+ Description : Stream merging tests+ Copyright : Ivan Lazar Miljenovic+ License : MIT+ Maintainer : Ivan.Miljenovic@gmail.com++++ -}+module Main (main) where++import Streaming.Concurrent++import Data.ByteString.Streaming (fromStrict, toStrict_)+import Streaming.Prelude (each, toList_)++import qualified Data.ByteString as B++import Data.Function (on)+import Data.List (concat, sort)+import Data.Monoid (mconcat)+import Test.Hspec (describe, hspec)+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck (Property, ioProperty)+import Test.QuickCheck.Instances ()++--------------------------------------------------------------------------------++main :: IO ()+main = hspec $ do+ describe "Merging retains all elements" $ do+ prop "Stream" (mergeCheck :: [[Int]] -> Property)+ prop "ByteString" mergeBSCheck++mergeCheck :: (Ord a) => [[a]] -> Property+mergeCheck ass = ioProperty (mergeStreams unbounded+ (map each ass)+ (eqOn sort as . toList_))+ where+ as = concat ass++mergeBSCheck :: [B.ByteString] -> Property+mergeBSCheck bss = ioProperty (mergeByteStrings unbounded+ (map fromStrict bss)+ (eqOn B.sort bs . toStrict_))+ where+ bs = mconcat bss++eqOn :: (Eq b, Functor f) => (a -> b) -> a -> f a -> f Bool+eqOn f a = fmap (((==)`on`f) a)