Running two pieces of Conduit in parallel (Neil Mitchell's blog)
neilmitchell.blogspot.co.uk/2015/07/parallelp...Posted Aug 3, 2015, 9:08 PM UTC
2611
neilmitchell.blogspot.co.uk
concurrent-machines offers various pipeline topologies with concurrent evaluation.
5
Thanks! http://hackage.haskell.org/package/concurrent-machines-0.1.0.0/docs/Data-Machine-Concurrent.html#v:buffer looks like my function.
2
I can't respond in detail due to being in the middle of traveling, but I think stm-conduit covers some or all of this. I haven't looked at that package in a while, but if it's missing since stuff, perhaps adding the functions there would make sense
5
I think http://haddock.stackage.org/nightly-2015-08-04/stm-conduit-2.6.1/Data-Conduit-Async.html#v:buffer is probably the function I've written (although the docs don't expose the constraints, so I can't confirm that my values are instances, but I suspect they are).
3
I'm probably not following, but how different is this problem from one typical for pipes-concurrency?
import Control.Monad
import Control.Concurrent.Async
import Control.Concurrent hiding (yield)
import Pipes.Concurrent
import qualified Pipes.Prelude as P
import Pipes
main = do
(output, input, seal) <- spawn' $ bounded 5
b <- async $ do runEffect $ pretend_parser >-> P.take 10 >-> toOutput output
atomically seal
a <- async $ do runEffect $ fromInput input >-> slow_worker
atomically seal
waitBoth a b
putStrLn "All done!"
pretend_parser = for (each [1..]) $ \n ->
do lift $ threadDelay 20000
lift $ putStrLn $ "sending " ++ show n ++ " to queue"
yield n
slow_worker = forever $ do
n <- await
lift $ do putStrLn $ show n ++ " removed from queue"
threadDelay 50000
putStrLn $ "*** work done on " ++ show n ++ " ***"
-- $ ./conc
-- sending 1 to queue
-- 1 removed from queue
-- sending 2 to queue
-- *** work done on 1 ***
-- sending 3 to queue
-- 2 removed from queue
-- sending 4 to queue
-- *** work done on 2 ***
-- sending 5 to queue
-- 3 removed from queue
-- sending 6 to queue
-- sending 7 to queue
-- *** work done on 3 ***
-- 4 removed from queue
-- sending 8 to queue
-- sending 9 to queue
-- *** work done on 4 ***
-- 5 removed from queue
-- sending 10 to queue
-- *** work done on 5 ***
-- 6 removed from queue
-- *** work done on 6 ***
-- 7 removed from queue
-- *** work done on 7 ***
-- 8 removed from queue
-- *** work done on 8 ***
-- 9 removed from queue
-- *** work done on 9 ***
-- 10 removed from queue
-- *** work done on 10 ***
-- All done!
4
The program above uses the style of the 'work stealing' examples from the tutorial. We can abstract something like this
buffer :: Int -> Producer a IO () -> Consumer a IO () -> IO ()
buffer n producer consumer = do
(output, input, seal) <- spawn' $ bounded n
b <- async $ do runEffect $ producer >-> toOutput output
atomically seal
a <- async $ do runEffect $ fromInput input >-> consumer
atomically seal
waitBoth a b >> return ()
-- a sort of buffered, concurrent version of Conduit.$$ but the machinery articulated in the tutorial admits all sorts of other arrangements and 'topologies'. In particular, the restriction to one consumer or producer is unnecessary.
For this use-case it would be clearer to use the names toBuffer and fromBuffer for toOutput and fromInput.
2
[deleted]
1
[deleted]
1
[deleted]
1
[deleted]
1
I couldn't find anything that does something similar in either pipes or conduits, but would appreciate any pointers.
I've implemented parI/parE for iteratees about 3-4 years ago. http://hackage.haskell.org/package/iteratee-0.8.9.6/docs/Data-Iteratee-Parallel.html
SCC has contained similar functionality based on pipeP for about six years, but I never got it to perform faster than the sequential version. Congratulations!
I'm not sure if it is possible, but consider generalizing the type to
parI/psequence_ helped to reduce processing time in one program by factor of 4...
I think MonadParallel probably is the magic sauce I need to generalise the monad - thanks! I'll probably leave it at IO only for my library, but nice to know about MonadParallel.