i've written small program which demonstrates how map/reduce may be implemented in Haskell. it counts amount of words in file, splitting it into 64kb blocks processed by two threads. their results are combined by another two threads. how it may be made better?
Hello, I do not know anything about
{-# LANGUAGE BangPatterns #-}
But your program logic certainly needs some polishing. Your idea is, to keep track of the number of thunks
jobs <- new 0 untilM (hIsEOF h) $ do str <- B.hGet h 65536 writeChan map str jobs += 1
jobs' <- val jobs writeChan reduce (0,-jobs')
and to collect partial sums, until all jobs are done.
reduceThread reduce result = forever $ do (sum1,n1) <- readChan reduce -- (*) (sum2,n2) <- readChan reduce let (!sum,!n) = (sum1+sum2,n1+n2) case n of 0 -> writeChan result sum _ -> writeChan reduce (sum,n)
I don't like the 'forever' here, especially since it prevents the program from re-usage. The real problem is the possibility of deadlocks: Suppose the situation, where all thunks are done, reducer A is directly before (*) and reducer B directly after (*). Now main sends the message (0,-jobs) and reducer A receives this message. Then both reducers are in state directly after (*) but no more messages are generated: A deadlock. The classical solution is to write an explicit END message into the channel and to keep track of the number of running mappers/reducers and not of the number of jobs. A simple attempt is mappers = 5 reducers = 3 main = do (file:_) <- getArgs h <- openBinaryFile file ReadMode map <- newChan red <- newChan res <- newChan replicateM_ mappers (forkIO $ mapThread map red) replicateM_ reducers (forkIO $ reduceThreadL red res 0) untilM (hIsEOF h) $ B.hGet h 65536 >>= writeChan map . Left writeChan map (Right mappers) -- explicit end-of-input message readChan result >>= print mapThread map red = readChan map >>= \ msg -> case msg of Left str -> writeChan red (Left $ length $ B.words str) >> mapThread map reduce Right 1 -> writeChan red (Right reducers) -- ^ explicit end-of-reduce message Right mappers_running -> writeChan map (Right $ mappers_running-1) reduceThreadL red result sum = readChan red >>= \ msg -> case msg of Left sum1 -> reduceThreadL red result (sum+sum1) -- ^ a new partial sum: reduce locally Right 1 -> writeChan res sum -- ^ I'm the last running reducer: produce the result now Right reducers_running -> writeChan red (Left sum) >> writeChan red (Right $ reducers_running-1) -- ^ reduce phase has ended, I'm not the last reducer: -- send partial reduction result and go home untilM cond action = cond >>= flip unless (action >> untilM cond action) It is quite easy to see that the program stops always, producing the correct result. There is no 'forever', so the program is re-usable. However, it is not exactly the same as the original program, since the local reduction results are stored locally and not send to the channel. If you want to do so, the problem becomes more complex. You have to solve the problem of distributed termination detection. There are algorithms for this, for example some simple token-based that need O(P) steps, where P is the number of processes involved. There are also some O(log P) algorithms, counting activation messages and summing up in two wave fronts these numbers. These algorithms are formulated for message passing environments with the capability to send a message to a specific receiver. I'm quite not sure how to adopt it to the channel-based environment (except via simulating the message passing environment). /BR, Mirko Rahn