abstract two Chan to Task

module Lib where

import Control.Concurrent.Chan

import Control.Concurrent (threadDelay)

import Data.Semigroup ((<>))

import Control.Monad (forever)

fetcher :: String -> IO Bool

fetcher url = do

  putStrLn $ "fetching " <> url

  threadDelay 1000000

  return $ if url `elem` ["url6", "url2"] then False else True

-- | submit actions to jobChan, worker will run it and put result in

-- resultChan.

worker :: Chan (IO a) -> Chan a -> IO ()

worker jobChan resultChan = forever $ do

  action <- readChan jobChan

  r <- action

  writeChan resultChan r

module Main where

import Control.Concurrent (threadDelay)

import Control.Concurrent.Async (Async, async, wait)

import Control.Monad (replicateM)

import Data.Semigroup ((<>))

import Control.Concurrent.QSem

import Control.Concurrent.Chan

fetcher :: String -> IO Bool

fetcher url = do

  putStrLn $ "fetching " <> url

  threadDelay 1000000

  return $ if url `elem` ["url6", "url2"] then False else True

-- | parallel map, run action using async.

pmapM :: (a -> IO b) -> [a] -> IO [Async b]

pmapM action lst = mapM (async . action) lst

-- | parallel map, run action using async, but max concurrency is guard by

-- QSem.

pmapSemM :: QSem -> (a -> IO b) -> [a] -> IO [Async b]

pmapSemM sem action lst =

    mapM (\a -> async $ do

                  waitQSem sem

                  action a

         ) lst

-- | each fetcher use its own thread.

downloadAll :: IO Bool

downloadAll = do

  let urlBatch1 = ["url1", "url2", "url3"]

  result1 :: [Async Bool] <- pmapM fetcher urlBatch1

  let urlBatch2 = ["url4", "url5", "url6", "url7", "url8"]

  result2 :: [Async Bool] <- pmapM fetcher urlBatch2

  result <- sequence $ map wait (result1 <> result2)

  let failedCount = length (filter (not . id) result)

  putStrLn $ show failedCount <> " failed"

  return $ and result

downloadAll2 :: IO Bool

downloadAll2 = do

  sem <- newQSem 2

  chan <- newChan :: IO (Chan (Async Bool))

  let urlBatch1 = ["url1", "url2", "url3"]

  result1 <- pmapSemM sem fetcher urlBatch1

  mapM_ (writeChan chan) result1

  let urlBatch2 = ["url4", "url5", "url6", "url7", "url8"]

  result2 <- pmapSemM sem fetcher urlBatch2

  mapM_ (writeChan chan) result2

  let expectedResultCount = length (urlBatch1 <> urlBatch2)

  results <- replicateM expectedResultCount

            (do

              as <- readChan chan

              b <- wait as

              signalQSem sem

              return b)

  let failedCount = length (filter (not . id) results)

  putStrLn $ show failedCount <> " failed"

  return $ and results

downloadAll3 :: IO Bool

downloadAll3 = do

  sem <- newQSem 2

  let urlBatch1 = ["url1", "url2", "url3"]

  result1 <- pmapSemM sem fetcher urlBatch1

  let urlBatch2 = ["url4", "url5", "url6", "url7", "url8"]

  result2 <- pmapSemM sem fetcher urlBatch2

  let asyncResults = result1 <> result2

  results <- mapM (\as -> do

                     b <- wait as

                     signalQSem sem

                     return b) asyncResults

  let failedCount = length (filter (not . id) results)

  putStrLn $ show failedCount <> " failed"

  return $ and results

import UseTwoChans (downloadAll)

import UseTask (downloadAll2)

main :: IO ()

main = do

  putStrLn "downloadAll2"

  result2 <- downloadAll2

  print result2

  putStrLn "downloadAll3"

  result3 <- downloadAll3

  print result3

module Task ( Task

            , newTask

            , addTask

            , addTasks

            , getTaskResults

            , ) where

import Control.Concurrent (forkIO)

import Control.Monad (replicateM, replicateM_, forever)

import Control.Concurrent.MVar

import Control.Concurrent.Chan

import Control.Exception

-- | submit actions to jobChan, worker will run it and put result in

-- resultChan.

worker :: Chan (IO a) -> Chan a -> IO ()

worker jobChan resultChan = forever $ do

  action <- readChan jobChan

  r <- action

  writeChan resultChan r

data Task a = Task {

      taskCount :: MVar Int

    , taskClosed :: MVar Bool

    , taskJobChan :: Chan (IO a)

    , taskResultChan :: Chan a }

data TaskException = TaskClosed deriving Show

instance Exception TaskException

-- | create a new task runner. TODO how to terminate workers when all job has

-- finished and task is closed?

newTask :: Int -> IO (Task a)

newTask n = do

  taskCountMVar <- newMVar 0

  taskClosedMVar <- newMVar False

  chan1 <- newChan

  chan2 <- newChan

  let result = Task {

                 taskCount=taskCountMVar

               , taskClosed=taskClosedMVar

               , taskJobChan=chan1

               , taskResultChan=chan2

               }

  replicateM_ n $ forkIO $ worker (taskJobChan result) (taskResultChan result)

  return result

-- | add action to task

addTask :: Task a -> IO a -> IO ()

addTask task action = do

  let tclosed = taskClosed task

  let tcount = taskCount task

  closed <- takeMVar tclosed

  if closed then do

      putMVar tclosed True

      throwIO TaskClosed

  else do

      writeChan (taskJobChan task) action

      count <- takeMVar tcount

      putMVar tcount (count + 1)

      putMVar tclosed False

-- | add a list of actions to task

addTasks :: Task a -> [IO a] -> IO ()

addTasks task actions = do

  let tclosed = taskClosed task

  let tcount = taskCount task

  closed <- takeMVar tclosed

  if closed then do

      putMVar tclosed True

      throwIO TaskClosed

  else do

      writeList2Chan (taskJobChan task) actions

      count <- takeMVar tcount

      putMVar tcount (count + length actions)

      putMVar tclosed False

-- | this is a blocking get. it will wait for all tasks to finish and return

-- result. this will also mark the Task as closed so no new task can be pushed

-- to it.

getTaskResults :: Task a -> IO [a]

getTaskResults task = do

  let tclosed = taskClosed task

  closed <- takeMVar tclosed  -- don't allow add new task when getTaskResults

                              -- is called.

  if closed then do

      putMVar tclosed True

      throwIO TaskClosed    -- can't run getTaskResults twice.

  else do

      putMVar tclosed True

      n <- readMVar (taskCount task)

      replicateM n (readChan (taskResultChan task))

module UseTask (downloadAll2) where

import Data.Semigroup ((<>))

import Lib (fetcher)

import Task

downloadAll2 :: IO Bool

downloadAll2 = do

  downloadTask <- newTask 5

  addTask downloadTask $ fetcher "url0"

  let urlBatch1 = ["url1", "url2", "url3"]

  addTasks downloadTask $ map fetcher urlBatch1

  let urlBatch2 = ["url4", "url5", "url6", "url7", "url8"]

  addTasks downloadTask $ map fetcher urlBatch2

  let urlBatch3 = ["url9", "url10", "url11", "url12", "url13", "url14"]

  addTasks downloadTask $ map fetcher urlBatch3

  results <- getTaskResults downloadTask

  let failedCount = length (filter (not . id) results)

  putStrLn $ show failedCount <> " failed"

  return $ and results

module UseTwoChans (downloadAll) where

import Control.Concurrent (forkIO)

import Control.Monad (replicateM, replicateM_)

import Control.Concurrent.Chan

import Data.Semigroup ((<>))

import Lib

downloadAll :: IO Bool

downloadAll = do

  jobChan :: Chan (IO Bool) <- newChan

  resultChan :: Chan Bool <- newChan

  replicateM_ 2 $ forkIO $ worker jobChan resultChan

  let urlBatch1 = ["url1", "url2", "url3"]

  mapM_ (writeChan jobChan . fetcher) urlBatch1

  let urlBatch2 = ["url4", "url5", "url6", "url7", "url8"]

  mapM_ (writeChan jobChan . fetcher) urlBatch2

  let urlBatch3 = ["url9", "url10", "url11", "url12", "url13", "url14"]

  mapM_ (writeChan jobChan . fetcher) urlBatch3

  let expectedResultCount = length $ urlBatch1 <> urlBatch2 <> urlBatch3

  results <- replicateM expectedResultCount (readChan resultChan)

  let failedCount = length (filter (not . id) results)

  putStrLn $ show failedCount <> " failed"

  return $ and results