diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,5 @@ +# Revision history for streaming-concurrency + +## 0.1.0.0 -- 2017-07-04 + +* First version. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,23 @@ +streaming-concurrency +============== + +[![Hackage](https://img.shields.io/hackage/v/streaming-concurrency.svg)](https://hackage.haskell.org/package/streaming-concurrency) [![Build Status](https://travis-ci.org/ivan-m/streaming-concurrency.svg)](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 diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/src/Streaming/Concurrent.hs b/src/Streaming/Concurrent.hs new file mode 100644 --- /dev/null +++ b/src/Streaming/Concurrent.hs @@ -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 #-} diff --git a/src/Streaming/Concurrent/Lifted.hs b/src/Streaming/Concurrent/Lifted.hs new file mode 100644 --- /dev/null +++ b/src/Streaming/Concurrent/Lifted.hs @@ -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) diff --git a/streaming-concurrency.cabal b/streaming-concurrency.cabal new file mode 100644 --- /dev/null +++ b/streaming-concurrency.cabal @@ -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 diff --git a/test/merging.hs b/test/merging.hs new file mode 100644 --- /dev/null +++ b/test/merging.hs @@ -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)