Commit a8183477 authored by Yuanle Song's avatar Yuanle Song
Browse files

implemented my TaskQueue, try3 works.

parent d2764e6f

Queue.hs

0 → 100644
+40 −0
Original line number Diff line number Diff line 
module Queue where



import Control.Monad

import Control.Concurrent.MVar

import Control.Concurrent



data TaskQueue a = TaskQueue {

      taskChan :: Chan a

    , taskCount :: MVar Int

    , taskQSem :: QSem

    }



-- | create a new Queue

newQueue :: IO (TaskQueue a)

newQueue = do

  ch <- newChan

  count <- newMVar 0

  sem <- newQSem 0

  return $ TaskQueue { taskChan=ch, taskCount=count, taskQSem=sem }



-- | add element to a Queue, block when size is full.

writeQueue :: TaskQueue a -> a -> IO ()

writeQueue queue item = do

  writeChan (taskChan queue) item

  modifyMVar_ (taskCount queue) (return . (+1))



-- | get element from a Queue, block when queue is empty.

readQueue :: TaskQueue a -> IO a

readQueue queue = readChan (taskChan queue)



-- | when get element from Queue, this mark the element is processed

-- successfully.

taskDone :: TaskQueue a -> IO ()

taskDone queue = signalQSem (taskQSem queue)



-- | block until all elements in Queue are processed successfully.

joinQueue :: TaskQueue a -> IO ()

joinQueue queue = do

  count <- readMVar (taskCount queue)

  replicateM_ count (waitQSem (taskQSem queue))

queue-base-threading.cabal

+17 −0
Original line number Diff line number Diff line
@@ -33,6 +33,23 @@ executable try2 
                     , random

  default-language:    Haskell2010



executable try3

  main-is:             try3.hs

  other-modules:       Queue

  ghc-options:         -threaded -rtsopts -with-rtsopts=-N

  build-depends:       base

                     , random

  default-language:    Haskell2010



executable testQueue

  main-is:             testQueue.hs

  other-modules:       Queue

  ghc-options:         -threaded -rtsopts -with-rtsopts=-N

  build-depends:       base

                     , random

  default-language:    Haskell2010





source-repository head

  type:     git

  location: https://github.com/sylecn/queue-base-threading

testQueue.hs

0 → 100644
+33 −0
Original line number Diff line number Diff line 
import Control.Concurrent

import Queue



test1 :: IO ()

test1 = do

  q <- newQueue

  writeQueue q 1

  writeQueue q 2

  writeQueue q 3

  taskDone q

  taskDone q

  taskDone q

  joinQueue q

  putStrLn "test1 done."



test2 :: IO ()

test2 = do

  q <- newQueue

  writeQueue q 1

  writeQueue q 2

  writeQueue q 3

  taskDone q

  taskDone q

  forkIO $ do

          threadDelay (5 * 1000000)

          taskDone q

  joinQueue q

  putStrLn "test2 done."



main :: IO ()

main = do

  test1

  test2

try2.hs

+0 −1
Original line number Diff line number Diff line
@@ -40,5 +40,4 @@ main = do 
  -- join counterCh

  writeChan printerCh ["Finishing up"]

  -- join printerCh

  -- done

  threadDelay $ 20 * 1000000

try3.hs

0 → 100644
+50 −0
Original line number Diff line number Diff line 
import Control.Monad

import Control.Concurrent.MVar

import Control.Concurrent

import Text.Printf

import System.Random



import Queue



fuzzDelay = False



fuzz :: IO ()

fuzz = if fuzzDelay

         then do

           seconds <- (getStdRandom random :: IO Double)

           threadDelay $ round (realToFrac (1000000 * seconds))

         else return ()



counterManager :: MVar Integer -> TaskQueue Integer -> TaskQueue [String] -> IO ()

counterManager counter counterQueue printerQueue = forever $ do

  fuzz

  incrementBy <- readQueue counterQueue

  fuzz

  newCount <- modifyMVar counter (\i -> return (i + 1, i + 1))

  fuzz

  writeQueue printerQueue [printf "The count is %d" newCount,

                           "---------------"]

  fuzz

  taskDone counterQueue



printerManager :: TaskQueue [String] -> IO ()

printerManager queue = forever $ do

  fuzz

  rows <- readQueue queue

  fuzz

  mapM_ putStrLn rows

  fuzz

  taskDone queue



main :: IO ()

main = do

  counter <- newMVar 0

  counterQueue <- newQueue

  printerQueue <- newQueue

  forkIO $ counterManager counter counterQueue printerQueue

  forkIO $ printerManager printerQueue

  writeQueue printerQueue ["Starting up"]

  replicateM_ 10 (writeQueue counterQueue 1)

  joinQueue counterQueue

  writeQueue printerQueue ["Finishing up"]

  joinQueue printerQueue