chan 0.0.0 → 0.0.1

raw patch · 3 files changed

+130/−1 lines, 3 filesdep +stmPVP ok

version bump matches the API change (PVP)

Dependencies added: stm

API changes (from Hackage documentation)

Files

• chan.cabal +4/−1
• src/Control/Concurrent/STM/TChan/Extra.hs +77/−0
• test/Spec.hs +49/−0

@@ -1,5 +1,5 @@ name:                chan-version:             0.0.0+version:             0.0.1 synopsis:            Some extra kit for Chans -- description: homepage:            https://github.com/athanclark/chan#readme@@ -16,8 +16,10 @@ library   hs-source-dirs:      src   exposed-modules:     Control.Concurrent.Chan.Extra+                       Control.Concurrent.STM.TChan.Extra   build-depends:       base >= 4.7 && < 5                      , async+                     , stm                      -- , since   default-language:    Haskell2010 @@ -28,6 +30,7 @@   build-depends:       base                      , chan                      , async+                     , stm   ghc-options:         -threaded -rtsopts -with-rtsopts=-N   default-language:    Haskell2010 

@@ -0,0 +1,77 @@+module Control.Concurrent.STM.TChan.Extra where++import Control.Monad (forever)+import Control.Concurrent (threadDelay)+import Control.Concurrent.STM (STM, atomically)+-- import Control.Concurrent.STM.TVar (newTVar, readIORef, writeIORef)+import Control.Concurrent.STM.TMVar (newEmptyTMVar, tryTakeTMVar, putTMVar)+import Control.Concurrent.STM.TChan (TChan, readTChan, writeTChan, newTChan)+import Control.Concurrent.Async (Async, async, cancel, wait)++++type DiffNanosec = Int+++-- | Note: In this model, even though we are using STM, a write to the+-- outgoing channel does not imply a transactional write to the output+-- channel; they are separated between a run IO layer, which means+-- we cannot atomically debounce or interleave the system (because+-- that depends on real-world time).+debounceStatic :: DiffNanosec -> TChan a -> IO (TChan a, Async ())+debounceStatic toWaitFurther outputChan = do+  (presentedChan,writingThread) <- atomically $ (,)+                                             <$> newTChan+                                             <*> newEmptyTMVar++  let invokeWrite x = do+        threadDelay toWaitFurther+        atomically $ writeTChan outputChan x++  writer <- async $ forever $ do+    x <- atomically $ readTChan presentedChan++    newWriter <- async (invokeWrite x)++    mInvoker <- atomically $ tryTakeTMVar writingThread+    case mInvoker of+      Nothing -> pure ()+      Just i -> cancel i+    atomically $ putTMVar writingThread newWriter++  pure (presentedChan, writer)++++intersperseStatic :: DiffNanosec -> IO a -> TChan a -> IO (TChan a, Async (), Async ())+intersperseStatic timeBetween xM outputChan = do+  (presentedChan,writingThread) <- atomically $ (,)+                                             <$> newTChan+                                             <*> newEmptyTMVar++  let invokeWritePing = do+        threadDelay timeBetween+        x <- xM+        atomically $ writeTChan outputChan x++  writer <- async $ forever $ do+    mInvoker <- atomically $ tryTakeTMVar writingThread+    case mInvoker of+      Nothing -> pure ()+      Just i -> wait i+    newWriter <- async invokeWritePing+    atomically $ putTMVar writingThread newWriter++  listener <- async $ forever $ do+    (y,mInvoker) <- atomically $ do+      y' <- readTChan presentedChan++      (\q -> (y',q)) <$> tryTakeTMVar writingThread++    case mInvoker of+      Nothing -> pure ()+      Just i -> cancel i++    atomically $ writeTChan outputChan y++  pure (presentedChan, writer, listener)

@@ -4,6 +4,10 @@ import Control.Concurrent.Chan.Extra (debounceStatic, intersperseStatic) import Control.Concurrent.Async (async) import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TChan (newTChan, writeTChan, readTChan)+import qualified Control.Concurrent.STM.TChan.Extra as TChan+import Control.Concurrent.STM.TMVar (newEmptyTMVar, putTMVar, takeTMVar)  main :: IO () main = do@@ -49,5 +53,50 @@   putStrLn "writing 2..."   writeChan incoming 2   _ <- takeMVar lock++  pure ()+++  outgoing <- atomically newTChan+  (incoming, _) <- TChan.debounceStatic 1000000 outgoing++  lock <- atomically newEmptyTMVar++  _ <- async $ forever $ do+    x <- atomically $ readTChan outgoing+    print x+    atomically $ putTMVar lock ()++  putStrLn "writing 1..."+  atomically $ writeTChan incoming 1++  _ <- atomically $ takeTMVar lock++  putStrLn "writing 2..."+  atomically $ writeTChan incoming 1+  atomically $ writeTChan incoming 2++  _ <- atomically $ takeTMVar lock++++  outgoing <- atomically newTChan+  (incoming, _, _) <- TChan.intersperseStatic 1000000 (pure 0) outgoing++  lock <- atomically newEmptyTMVar++  _ <- async $ forever $ do+    x <- atomically $ readTChan outgoing+    print x+    atomically $ putTMVar lock ()++  threadDelay 3000000++  putStrLn "Writing 1..."+  atomically $ writeTChan incoming 1+  _ <- atomically $ takeTMVar lock+  putStrLn "Writing 2..."+  atomically $ writeTChan incoming 2+  _ <- atomically $ takeTMVar lock    pure ()