@@ -23,13 +23,12 @@ module Control.AutoUpdate (
2323 , mkAutoUpdate
2424 ) where
2525
26−import Control.AutoUpdate.Util (atomicModifyIORef')
27−import Control.Concurrent (ThreadId, forkIO, myThreadId, threadDelay)
28−import Control.Exception (Exception, SomeException
29− ,assert, fromException, handle,throwIO, throwTo)
30−import Control.Monad (forever, join)
31−import Data.IORef (IORef, newIORef)
32−import Data.Typeable (Typeable)
26+import Control.Concurrent (forkIO, threadDelay)
27+import Control.Concurrent.MVar (newEmptyMVar, putMVar, readMVar,
28+ takeMVar, tryPutMVar, tryTakeMVar)
29+import Control.Exception (SomeException, catch, throw)
30+import Control.Monad (forever, void)
31+import Data.IORef (newIORef, readIORef, writeIORef)
3332
3433-- | Default value for creating an @UpdateSettings@.
3534--
@@ -60,8 +59,11 @@ data UpdateSettings a = UpdateSettings
6059 --
6160 -- Since 0.1.0
6261 , updateSpawnThreshold :: Int
63− -- ^ How many times the data must be requested before we decide to
64− -- spawn a dedicated thread.
62+ -- ^ NOTE: This value no longer has any effect, since worker threads are
63+ -- dedicated instead of spawned on demand.
64+ --
65+ -- Previously, this determined: How many times the data must be requested
66+ -- before we decide to spawn a dedicated thread.
6567 --
6668 -- Default: 3
6769 --
@@ -74,94 +76,57 @@ data UpdateSettings a = UpdateSettings
7476 -- Since 0.1.0
7577 }
7678
77−data Status a = AutoUpdated
78− !a
79− {-# UNPACK #-} !Int
80− -- Number of times used since last updated.
81− {-# UNPACK #-} !ThreadId
82− -- Worker thread.
83− | ManualUpdates
84− {-# UNPACK #-} !Int
85− -- Number of times used since we started/switched
86− -- off manual updates.
87−
8879-- | Generate an action which will either read from an automatically
8980-- updated value, or run the update action in the current thread.
9081--
9182-- Since 0.1.0
9283mkAutoUpdate :: UpdateSettings a -> IO (IO a)
9384mkAutoUpdate us = do
94− istatus <- newIORef $ ManualUpdates 0
95− return $! getCurrent us istatus
85+ -- The current value, if available.
86+ currRef <- newIORef Nothing
9687
97−data Action a = Return a | Manual | Spawn
88+ -- A baton to tell the worker thread to generate a new value.
89+ needsRunning <- newEmptyMVar
9890
99−data Replaced = Replaced deriving (Show, Typeable)
100−instance Exception Replaced
91+ -- The last value generated, to allow for blocking semantics when currRef
92+ -- is Nothing.
93+ lastValue <- newEmptyMVar
10194
102−-- | Get the current value, either fed from an auto-update thread, or
103−-- computed manually in the current thread.
104−--
105−-- Since 0.1.0
106−getCurrent :: UpdateSettings a
107− -> IORef (Status a) -- ^ mutable state
108− -> IO a
109−getCurrent settings@UpdateSettings{..} istatus = do
110− ea <- atomicModifyIORef' istatus increment
111− case ea of
112− Return a -> return a
113− Manual -> updateAction
114− Spawn -> do
115− a <- updateAction
116− tid <- forkIO $ spawn settings istatus
117− join $ atomicModifyIORef' istatus $ turnToAuto a tid
118− return a
119− where
120− increment (AutoUpdated a cnt tid) = (AutoUpdated a (succ cnt) tid, Return a)
121− increment (ManualUpdates i) = (ManualUpdates (succ i), act)
122− where
123− act = if i > updateSpawnThreshold then Spawn else Manual
95+ -- fork the worker thread immediately...
96+ void $ forkIO $ forever $ do
97+ -- but block until a value is actually needed
98+ takeMVar needsRunning
99+
100+ -- new value requested, so run the updateAction
101+ a <- catchSome $ updateAction us
102+
103+ -- we got a new value, update currRef and lastValue
104+ writeIORef currRef $ Just a
105+ void $ tryTakeMVar lastValue
106+ putMVar lastValue a
107+
108+ -- delay until we're needed again
109+ threadDelay $ updateFreq us
124110
125− -- Normal case.
126− turnToAuto a tid (ManualUpdates cnt) = (AutoUpdated a cnt tid
127− ,return ())
128− -- Race condition: multiple threads were spawned.
129− -- So, let's kill the previous one by this thread.
130− turnToAuto a tid (AutoUpdated _ cnt old) = (AutoUpdated a cnt tid
131− ,throwTo old Replaced)
111+ -- delay's over, clear out currRef and lastValue so that demanding the
112+ -- value again forces us to start work
113+ writeIORef currRef Nothing
114+ void $ takeMVar lastValue
132115
133−spawn :: UpdateSettings a -> IORef (Status a) -> IO ()
134−spawn UpdateSettings{..} istatus = handle (onErr istatus) $ forever $ do
135− threadDelay updateFreq
136− a <- updateAction
137− join $ atomicModifyIORef' istatus $ turnToManual a
138− where
139− -- Normal case.
140− turnToManual a (AutoUpdated _ cnt tid)
141− | cnt >= 1 = (AutoUpdated a 0 tid, return ())
142− | otherwise = (ManualUpdates 0, stop)
143− -- This case must not happen.
144− turnToManual _ (ManualUpdates i) = assert False (ManualUpdates i, stop)
116+ return $ do
117+ mval <- readIORef currRef
118+ case mval of
119+ -- we have a current value, use it
120+ Just val -> return val
121+ Nothing -> do
122+ -- no current value, force the worker thread to run...
123+ void $ tryPutMVar needsRunning ()
145124
146−onErr :: IORef (Status a) -> SomeException -> IO ()
147−onErr istatus ex = case fromException ex of
148− Just Replaced -> return () -- this thread is terminated
149− Nothing -> do
150− tid <- myThreadId
151− atomicModifyIORef' istatus $ clear tid
152− throwIO ex
153− where
154− -- In the race condition described above,
155− -- suppose thread A is running, and is killed by thread B.
156− -- Thread B then updates the IORef to refer to thread B.
157− -- Then thread A's exception handler fires.
158− -- We don't want to modify the IORef at all,
159− -- since it refers to thread B already.
160− -- Solution: only switch back to manual updates
161− -- if the IORef is pointing at the current thread.
162− clear tid (AutoUpdated _ _ tid') | tid == tid' = (ManualUpdates 0, ())
163− clear _ status = (status, ())
125+ -- and block for the result from the worker
126+ readMVar lastValue
164127
165−-- | Throw an error to kill a thread.
166−stop :: IO a
167−stop = throwIO Replaced
128+-- | Turn a runtime exception into an impure exception, so that all @IO@
129+-- actions will complete successfully. This simply defers the exception until
130+-- the value is forced.
131+catchSome :: IO a -> IO a
132+catchSome act = Control.Exception.catch act $ \e -> return $ throw (e :: SomeException)