Commit 98da59a1 authored by Yuanle Song's avatar Yuanle Song
Browse files

allow submit task after calling getTaskResults

parent a87cb6cb

app/UseTask.hs

+9 −2
Original line number Diff line number Diff line
@@ -13,10 +13,17 @@ downloadAll2 = do 
  addTasks downloadTask $ map fetcher urlBatch1

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

  addTasks downloadTask $ map fetcher urlBatch2



  results1 <- getTaskResults downloadTask



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

  addTasks downloadTask $ map fetcher urlBatch3



  results <- getTaskResults downloadTask

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

  results2 <- getTaskResults downloadTask



  let results = results1 <> results2

      totalCount = length results

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

  putStrLn $ show totalCount <> " task ran"

  putStrLn $ show failedCount <> " failed"

  return $ and results

package.yaml

+6 −1
Original line number Diff line number Diff line 
name:                io-thread-pool

version:             0.1.0.0

version:             0.2.0.0

github:              "sylecn/io-thread-pool"

license:             BSD3

author:              "Yuanle Song"
@@ -31,6 +31,9 @@ dependencies: 
- base >= 4.7 && < 5

- async



library:

  source-dirs: src



executables:

  t1:

    main:                Main.hs
@@ -39,3 +42,5 @@ executables: 
    - -threaded

    - -rtsopts

    - -with-rtsopts=-N

    dependencies:

    - io-thread-pool

app/Task.hs→src/Task.hs

+5 −29
Original line number Diff line number Diff line
@@ -21,7 +21,6 @@ worker jobChan resultChan = forever $ do 


data Task a = Task {

      taskCount :: MVar Int

    , taskClosed :: MVar Bool

    , taskJobChan :: Chan (IO a)

    , taskResultChan :: Chan a }
@@ -34,12 +33,10 @@ instance Exception TaskException 
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

               }
@@ -49,45 +46,24 @@ newTask n = do 
-- | 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.

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

-- and return result. if new tasks are submitted, you can call getTaskResults

-- again.

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)

  n <- takeMVar (taskCount task)

  putMVar (taskCount task) 0

  replicateM n (readChan (taskResultChan task))