Use explicit language extensions & remove extension fields from base.cabal
[packages/base.git] / System / Event / Manager.hs
1 {-# LANGUAGE BangPatterns
2 , CPP
3 , ExistentialQuantification
4 , NoImplicitPrelude
5 , RecordWildCards
6 , TypeSynonymInstances
7 , FlexibleInstances
8 #-}
9
10 module System.Event.Manager
11 ( -- * Types
12 EventManager
13
14 -- * Creation
15 , new
16 , newWith
17 , newDefaultBackend
18
19 -- * Running
20 , finished
21 , loop
22 , step
23 , shutdown
24 , cleanup
25 , wakeManager
26
27 -- * Registering interest in I/O events
28 , Event
29 , evtRead
30 , evtWrite
31 , IOCallback
32 , FdKey(keyFd)
33 , registerFd_
34 , registerFd
35 , unregisterFd_
36 , unregisterFd
37 , closeFd
38
39 -- * Registering interest in timeout events
40 , TimeoutCallback
41 , TimeoutKey
42 , registerTimeout
43 , updateTimeout
44 , unregisterTimeout
45 ) where
46
47 #include "EventConfig.h"
48
49 ------------------------------------------------------------------------
50 -- Imports
51
52 import Control.Concurrent.MVar (MVar, modifyMVar, newMVar, readMVar)
53 import Control.Exception (finally)
54 import Control.Monad ((=<<), forM_, liftM, sequence_, when)
55 import Data.IORef (IORef, atomicModifyIORef, mkWeakIORef, newIORef, readIORef,
56 writeIORef)
57 import Data.Maybe (Maybe(..))
58 import Data.Monoid (mappend, mconcat, mempty)
59 import GHC.Base
60 import GHC.Conc.Signal (runHandlers)
61 import GHC.List (filter)
62 import GHC.Num (Num(..))
63 import GHC.Real ((/), fromIntegral )
64 import GHC.Show (Show(..))
65 import System.Event.Clock (getCurrentTime)
66 import System.Event.Control
67 import System.Event.Internal (Backend, Event, evtClose, evtRead, evtWrite,
68 Timeout(..))
69 import System.Event.Unique (Unique, UniqueSource, newSource, newUnique)
70 import System.Posix.Types (Fd)
71
72 import qualified System.Event.IntMap as IM
73 import qualified System.Event.Internal as I
74 import qualified System.Event.PSQ as Q
75
76 #if defined(HAVE_KQUEUE)
77 import qualified System.Event.KQueue as KQueue
78 #elif defined(HAVE_EPOLL)
79 import qualified System.Event.EPoll as EPoll
80 #elif defined(HAVE_POLL)
81 import qualified System.Event.Poll as Poll
82 #else
83 # error not implemented for this operating system
84 #endif
85
86 ------------------------------------------------------------------------
87 -- Types
88
89 data FdData = FdData {
90 fdKey :: {-# UNPACK #-} !FdKey
91 , fdEvents :: {-# UNPACK #-} !Event
92 , _fdCallback :: !IOCallback
93 } deriving (Show)
94
95 -- | A file descriptor registration cookie.
96 data FdKey = FdKey {
97 keyFd :: {-# UNPACK #-} !Fd
98 , keyUnique :: {-# UNPACK #-} !Unique
99 } deriving (Eq, Show)
100
101 -- | Callback invoked on I/O events.
102 type IOCallback = FdKey -> Event -> IO ()
103
104 instance Show IOCallback where
105 show _ = "IOCallback"
106
107 -- | A timeout registration cookie.
108 newtype TimeoutKey = TK Unique
109 deriving (Eq)
110
111 -- | Callback invoked on timeout events.
112 type TimeoutCallback = IO ()
113
114 data State = Created
115 | Running
116 | Dying
117 | Finished
118 deriving (Eq, Show)
119
120 -- | A priority search queue, with timeouts as priorities.
121 type TimeoutQueue = Q.PSQ TimeoutCallback
122
123 {-
124 Instead of directly modifying the 'TimeoutQueue' in
125 e.g. 'registerTimeout' we keep a list of edits to perform, in the form
126 of a chain of function closures, and have the I/O manager thread
127 perform the edits later. This exist to address the following GC
128 problem:
129
130 Since e.g. 'registerTimeout' doesn't force the evaluation of the
131 thunks inside the 'emTimeouts' IORef a number of thunks build up
132 inside the IORef. If the I/O manager thread doesn't evaluate these
133 thunks soon enough they'll get promoted to the old generation and
134 become roots for all subsequent minor GCs.
135
136 When the thunks eventually get evaluated they will each create a new
137 intermediate 'TimeoutQueue' that immediately becomes garbage. Since
138 the thunks serve as roots until the next major GC these intermediate
139 'TimeoutQueue's will get copied unnecesarily in the next minor GC,
140 increasing GC time. This problem is known as "floating garbage".
141
142 Keeping a list of edits doesn't stop this from happening but makes the
143 amount of data that gets copied smaller.
144
145 TODO: Evaluate the content of the IORef to WHNF on each insert once
146 this bug is resolved: http://hackage.haskell.org/trac/ghc/ticket/3838
147 -}
148
149 -- | An edit to apply to a 'TimeoutQueue'.
150 type TimeoutEdit = TimeoutQueue -> TimeoutQueue
151
152 -- | The event manager state.
153 data EventManager = EventManager
154 { emBackend :: !Backend
155 , emFds :: {-# UNPACK #-} !(MVar (IM.IntMap [FdData]))
156 , emTimeouts :: {-# UNPACK #-} !(IORef TimeoutEdit)
157 , emState :: {-# UNPACK #-} !(IORef State)
158 , emUniqueSource :: {-# UNPACK #-} !UniqueSource
159 , emControl :: {-# UNPACK #-} !Control
160 }
161
162 ------------------------------------------------------------------------
163 -- Creation
164
165 handleControlEvent :: EventManager -> FdKey -> Event -> IO ()
166 handleControlEvent mgr reg _evt = do
167 msg <- readControlMessage (emControl mgr) (keyFd reg)
168 case msg of
169 CMsgWakeup -> return ()
170 CMsgDie -> writeIORef (emState mgr) Finished
171 CMsgSignal fp s -> runHandlers fp s
172
173 newDefaultBackend :: IO Backend
174 #if defined(HAVE_KQUEUE)
175 newDefaultBackend = KQueue.new
176 #elif defined(HAVE_EPOLL)
177 newDefaultBackend = EPoll.new
178 #elif defined(HAVE_POLL)
179 newDefaultBackend = Poll.new
180 #else
181 newDefaultBackend = error "no back end for this platform"
182 #endif
183
184 -- | Create a new event manager.
185 new :: IO EventManager
186 new = newWith =<< newDefaultBackend
187
188 newWith :: Backend -> IO EventManager
189 newWith be = do
190 iofds <- newMVar IM.empty
191 timeouts <- newIORef id
192 ctrl <- newControl
193 state <- newIORef Created
194 us <- newSource
195 _ <- mkWeakIORef state $ do
196 st <- atomicModifyIORef state $ \s -> (Finished, s)
197 when (st /= Finished) $ do
198 I.delete be
199 closeControl ctrl
200 let mgr = EventManager { emBackend = be
201 , emFds = iofds
202 , emTimeouts = timeouts
203 , emState = state
204 , emUniqueSource = us
205 , emControl = ctrl
206 }
207 _ <- registerFd_ mgr (handleControlEvent mgr) (controlReadFd ctrl) evtRead
208 _ <- registerFd_ mgr (handleControlEvent mgr) (wakeupReadFd ctrl) evtRead
209 return mgr
210
211 -- | Asynchronously shuts down the event manager, if running.
212 shutdown :: EventManager -> IO ()
213 shutdown mgr = do
214 state <- atomicModifyIORef (emState mgr) $ \s -> (Dying, s)
215 when (state == Running) $ sendDie (emControl mgr)
216
217 finished :: EventManager -> IO Bool
218 finished mgr = (== Finished) `liftM` readIORef (emState mgr)
219
220 cleanup :: EventManager -> IO ()
221 cleanup EventManager{..} = do
222 writeIORef emState Finished
223 I.delete emBackend
224 closeControl emControl
225
226 ------------------------------------------------------------------------
227 -- Event loop
228
229 -- | Start handling events. This function loops until told to stop,
230 -- using 'shutdown'.
231 --
232 -- /Note/: This loop can only be run once per 'EventManager', as it
233 -- closes all of its control resources when it finishes.
234 loop :: EventManager -> IO ()
235 loop mgr@EventManager{..} = do
236 state <- atomicModifyIORef emState $ \s -> case s of
237 Created -> (Running, s)
238 _ -> (s, s)
239 case state of
240 Created -> go Q.empty `finally` cleanup mgr
241 Dying -> cleanup mgr
242 _ -> do cleanup mgr
243 error $ "System.Event.Manager.loop: state is already " ++
244 show state
245 where
246 go q = do (running, q') <- step mgr q
247 when running $ go q'
248
249 step :: EventManager -> TimeoutQueue -> IO (Bool, TimeoutQueue)
250 step mgr@EventManager{..} tq = do
251 (timeout, q') <- mkTimeout tq
252 I.poll emBackend timeout (onFdEvent mgr)
253 state <- readIORef emState
254 state `seq` return (state == Running, q')
255 where
256
257 -- | Call all expired timer callbacks and return the time to the
258 -- next timeout.
259 mkTimeout :: TimeoutQueue -> IO (Timeout, TimeoutQueue)
260 mkTimeout q = do
261 now <- getCurrentTime
262 applyEdits <- atomicModifyIORef emTimeouts $ \f -> (id, f)
263 let (expired, q'') = let q' = applyEdits q in q' `seq` Q.atMost now q'
264 sequence_ $ map Q.value expired
265 let timeout = case Q.minView q'' of
266 Nothing -> Forever
267 Just (Q.E _ t _, _) ->
268 -- This value will always be positive since the call
269 -- to 'atMost' above removed any timeouts <= 'now'
270 let t' = t - now in t' `seq` Timeout t'
271 return (timeout, q'')
272
273 ------------------------------------------------------------------------
274 -- Registering interest in I/O events
275
276 -- | Register interest in the given events, without waking the event
277 -- manager thread. The 'Bool' return value indicates whether the
278 -- event manager ought to be woken.
279 registerFd_ :: EventManager -> IOCallback -> Fd -> Event
280 -> IO (FdKey, Bool)
281 registerFd_ EventManager{..} cb fd evs = do
282 u <- newUnique emUniqueSource
283 modifyMVar emFds $ \oldMap -> do
284 let fd' = fromIntegral fd
285 reg = FdKey fd u
286 !fdd = FdData reg evs cb
287 (!newMap, (oldEvs, newEvs)) =
288 case IM.insertWith (++) fd' [fdd] oldMap of
289 (Nothing, n) -> (n, (mempty, evs))
290 (Just prev, n) -> (n, pairEvents prev newMap fd')
291 modify = oldEvs /= newEvs
292 when modify $ I.modifyFd emBackend fd oldEvs newEvs
293 return (newMap, (reg, modify))
294 {-# INLINE registerFd_ #-}
295
296 -- | @registerFd mgr cb fd evs@ registers interest in the events @evs@
297 -- on the file descriptor @fd@. @cb@ is called for each event that
298 -- occurs. Returns a cookie that can be handed to 'unregisterFd'.
299 registerFd :: EventManager -> IOCallback -> Fd -> Event -> IO FdKey
300 registerFd mgr cb fd evs = do
301 (r, wake) <- registerFd_ mgr cb fd evs
302 when wake $ wakeManager mgr
303 return r
304 {-# INLINE registerFd #-}
305
306 -- | Wake up the event manager.
307 wakeManager :: EventManager -> IO ()
308 wakeManager mgr = sendWakeup (emControl mgr)
309
310 eventsOf :: [FdData] -> Event
311 eventsOf = mconcat . map fdEvents
312
313 pairEvents :: [FdData] -> IM.IntMap [FdData] -> Int -> (Event, Event)
314 pairEvents prev m fd = let l = eventsOf prev
315 r = case IM.lookup fd m of
316 Nothing -> mempty
317 Just fds -> eventsOf fds
318 in (l, r)
319
320 -- | Drop a previous file descriptor registration, without waking the
321 -- event manager thread. The return value indicates whether the event
322 -- manager ought to be woken.
323 unregisterFd_ :: EventManager -> FdKey -> IO Bool
324 unregisterFd_ EventManager{..} (FdKey fd u) =
325 modifyMVar emFds $ \oldMap -> do
326 let dropReg cbs = case filter ((/= u) . keyUnique . fdKey) cbs of
327 [] -> Nothing
328 cbs' -> Just cbs'
329 fd' = fromIntegral fd
330 (!newMap, (oldEvs, newEvs)) =
331 case IM.updateWith dropReg fd' oldMap of
332 (Nothing, _) -> (oldMap, (mempty, mempty))
333 (Just prev, newm) -> (newm, pairEvents prev newm fd')
334 modify = oldEvs /= newEvs
335 when modify $ I.modifyFd emBackend fd oldEvs newEvs
336 return (newMap, modify)
337
338 -- | Drop a previous file descriptor registration.
339 unregisterFd :: EventManager -> FdKey -> IO ()
340 unregisterFd mgr reg = do
341 wake <- unregisterFd_ mgr reg
342 when wake $ wakeManager mgr
343
344 -- | Close a file descriptor in a race-safe way.
345 closeFd :: EventManager -> (Fd -> IO ()) -> Fd -> IO ()
346 closeFd mgr close fd = do
347 fds <- modifyMVar (emFds mgr) $ \oldMap -> do
348 close fd
349 case IM.delete (fromIntegral fd) oldMap of
350 (Nothing, _) -> return (oldMap, [])
351 (Just fds, !newMap) -> do
352 when (eventsOf fds /= mempty) $ wakeManager mgr
353 return (newMap, fds)
354 forM_ fds $ \(FdData reg ev cb) -> cb reg (ev `mappend` evtClose)
355
356 ------------------------------------------------------------------------
357 -- Registering interest in timeout events
358
359 -- | Register a timeout in the given number of microseconds. The
360 -- returned 'TimeoutKey' can be used to later unregister or update the
361 -- timeout. The timeout is automatically unregistered after the given
362 -- time has passed.
363 registerTimeout :: EventManager -> Int -> TimeoutCallback -> IO TimeoutKey
364 registerTimeout mgr us cb = do
365 !key <- newUnique (emUniqueSource mgr)
366 if us <= 0 then cb
367 else do
368 now <- getCurrentTime
369 let expTime = fromIntegral us / 1000000.0 + now
370
371 -- We intentionally do not evaluate the modified map to WHNF here.
372 -- Instead, we leave a thunk inside the IORef and defer its
373 -- evaluation until mkTimeout in the event loop. This is a
374 -- workaround for a nasty IORef contention problem that causes the
375 -- thread-delay benchmark to take 20 seconds instead of 0.2.
376 atomicModifyIORef (emTimeouts mgr) $ \f ->
377 let f' = (Q.insert key expTime cb) . f in (f', ())
378 wakeManager mgr
379 return $ TK key
380
381 -- | Unregister an active timeout.
382 unregisterTimeout :: EventManager -> TimeoutKey -> IO ()
383 unregisterTimeout mgr (TK key) = do
384 atomicModifyIORef (emTimeouts mgr) $ \f ->
385 let f' = (Q.delete key) . f in (f', ())
386 wakeManager mgr
387
388 -- | Update an active timeout to fire in the given number of
389 -- microseconds.
390 updateTimeout :: EventManager -> TimeoutKey -> Int -> IO ()
391 updateTimeout mgr (TK key) us = do
392 now <- getCurrentTime
393 let expTime = fromIntegral us / 1000000.0 + now
394
395 atomicModifyIORef (emTimeouts mgr) $ \f ->
396 let f' = (Q.adjust (const expTime) key) . f in (f', ())
397 wakeManager mgr
398
399 ------------------------------------------------------------------------
400 -- Utilities
401
402 -- | Call the callbacks corresponding to the given file descriptor.
403 onFdEvent :: EventManager -> Fd -> Event -> IO ()
404 onFdEvent mgr fd evs = do
405 fds <- readMVar (emFds mgr)
406 case IM.lookup (fromIntegral fd) fds of
407 Just cbs -> forM_ cbs $ \(FdData reg ev cb) ->
408 when (evs `I.eventIs` ev) $ cb reg evs
409 Nothing -> return ()