2 {-# OPTIONS_GHC -XNoImplicitPrelude #-}
3 {-# OPTIONS_HADDOCK not-home #-}
4 -----------------------------------------------------------------------------
7 -- Copyright : (c) The University of Glasgow, 1994-2002
8 -- License : see libraries/base/LICENSE
10 -- Maintainer : cvs-ghc@haskell.org
11 -- Stability : internal
12 -- Portability : non-portable (GHC extensions)
14 -- Basic concurrency stuff.
16 -----------------------------------------------------------------------------
18 -- No: #hide, because bits of this module are exposed by the stm package.
19 -- However, we don't want this module to be the home location for the
20 -- bits it exports, we'd rather have Control.Concurrent and the other
21 -- higher level modules be the home. Hence:
29 -- * Forking and suchlike
30 , forkIO -- :: IO a -> IO ThreadId
31 , forkOnIO -- :: Int -> IO a -> IO ThreadId
32 , numCapabilities -- :: Int
33 , childHandler -- :: Exception -> IO ()
34 , myThreadId -- :: IO ThreadId
35 , killThread -- :: ThreadId -> IO ()
36 , throwTo -- :: ThreadId -> Exception -> IO ()
37 , par -- :: a -> b -> b
38 , pseq -- :: a -> b -> b
40 , labelThread -- :: ThreadId -> String -> IO ()
42 , ThreadStatus(..), BlockReason(..)
43 , threadStatus -- :: ThreadId -> IO ThreadStatus
46 , threadDelay -- :: Int -> IO ()
47 , registerDelay -- :: Int -> IO (TVar Bool)
48 , threadWaitRead -- :: Int -> IO ()
49 , threadWaitWrite -- :: Int -> IO ()
53 , newMVar -- :: a -> IO (MVar a)
54 , newEmptyMVar -- :: IO (MVar a)
55 , takeMVar -- :: MVar a -> IO a
56 , putMVar -- :: MVar a -> a -> IO ()
57 , tryTakeMVar -- :: MVar a -> IO (Maybe a)
58 , tryPutMVar -- :: MVar a -> a -> IO Bool
59 , isEmptyMVar -- :: MVar a -> IO Bool
60 , addMVarFinalizer -- :: MVar a -> IO () -> IO ()
64 , atomically -- :: STM a -> IO a
66 , orElse -- :: STM a -> STM a -> STM a
67 , catchSTM -- :: STM a -> (Exception -> STM a) -> STM a
68 , alwaysSucceeds -- :: STM a -> STM ()
69 , always -- :: STM Bool -> STM ()
71 , newTVar -- :: a -> STM (TVar a)
72 , newTVarIO -- :: a -> STM (TVar a)
73 , readTVar -- :: TVar a -> STM a
74 , writeTVar -- :: a -> TVar a -> STM ()
75 , unsafeIOToSTM -- :: IO a -> STM a
78 #ifdef mingw32_HOST_OS
79 , asyncRead -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
80 , asyncWrite -- :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
81 , asyncDoProc -- :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
83 , asyncReadBA -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
84 , asyncWriteBA -- :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int, Int)
87 #ifndef mingw32_HOST_OS
91 , ensureIOManagerIsRunning
93 #ifdef mingw32_HOST_OS
100 import System.Posix.Types
101 #ifndef mingw32_HOST_OS
102 import System.Posix.Internals
108 import {-# SOURCE #-} GHC.TopHandler ( reportError, reportStackOverflow )
115 import GHC.Num ( Num(..) )
116 import GHC.Real ( fromIntegral, div )
117 #ifndef mingw32_HOST_OS
118 import GHC.Base ( Int(..) )
120 #ifdef mingw32_HOST_OS
121 import GHC.Read ( Read )
122 import GHC.Enum ( Enum )
124 import GHC.Exception ( SomeException(..), throw )
125 import GHC.Pack ( packCString# )
126 import GHC.Ptr ( Ptr(..), plusPtr, FunPtr(..) )
128 import GHC.Show ( Show(..), showString )
131 infixr 0 `par`, `pseq`
134 %************************************************************************
136 \subsection{@ThreadId@, @par@, and @fork@}
138 %************************************************************************
141 data ThreadId = ThreadId ThreadId# deriving( Typeable )
142 -- ToDo: data ThreadId = ThreadId (Weak ThreadId#)
143 -- But since ThreadId# is unlifted, the Weak type must use open
146 A 'ThreadId' is an abstract type representing a handle to a thread.
147 'ThreadId' is an instance of 'Eq', 'Ord' and 'Show', where
148 the 'Ord' instance implements an arbitrary total ordering over
149 'ThreadId's. The 'Show' instance lets you convert an arbitrary-valued
150 'ThreadId' to string form; showing a 'ThreadId' value is occasionally
151 useful when debugging or diagnosing the behaviour of a concurrent
154 /Note/: in GHC, if you have a 'ThreadId', you essentially have
155 a pointer to the thread itself. This means the thread itself can\'t be
156 garbage collected until you drop the 'ThreadId'.
157 This misfeature will hopefully be corrected at a later date.
159 /Note/: Hugs does not provide any operations on other threads;
160 it defines 'ThreadId' as a synonym for ().
163 instance Show ThreadId where
165 showString "ThreadId " .
166 showsPrec d (getThreadId (id2TSO t))
168 foreign import ccall unsafe "rts_getThreadId" getThreadId :: ThreadId# -> CInt
170 id2TSO :: ThreadId -> ThreadId#
171 id2TSO (ThreadId t) = t
173 foreign import ccall unsafe "cmp_thread" cmp_thread :: ThreadId# -> ThreadId# -> CInt
176 cmpThread :: ThreadId -> ThreadId -> Ordering
178 case cmp_thread (id2TSO t1) (id2TSO t2) of
183 instance Eq ThreadId where
185 case t1 `cmpThread` t2 of
189 instance Ord ThreadId where
193 Sparks off a new thread to run the 'IO' computation passed as the
194 first argument, and returns the 'ThreadId' of the newly created
197 The new thread will be a lightweight thread; if you want to use a foreign
198 library that uses thread-local storage, use 'Control.Concurrent.forkOS' instead.
200 GHC note: the new thread inherits the /blocked/ state of the parent
201 (see 'Control.Exception.block').
203 forkIO :: IO () -> IO ThreadId
204 forkIO action = IO $ \ s ->
205 case (fork# action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
207 action_plus = catchException action childHandler
210 Like 'forkIO', but lets you specify on which CPU the thread is
211 created. Unlike a `forkIO` thread, a thread created by `forkOnIO`
212 will stay on the same CPU for its entire lifetime (`forkIO` threads
213 can migrate between CPUs according to the scheduling policy).
214 `forkOnIO` is useful for overriding the scheduling policy when you
215 know in advance how best to distribute the threads.
217 The `Int` argument specifies the CPU number; it is interpreted modulo
218 'numCapabilities' (note that it actually specifies a capability number
219 rather than a CPU number, but to a first approximation the two are
222 forkOnIO :: Int -> IO () -> IO ThreadId
223 forkOnIO (I# cpu) action = IO $ \ s ->
224 case (forkOn# cpu action_plus s) of (# s1, id #) -> (# s1, ThreadId id #)
226 action_plus = catchException action childHandler
228 -- | the value passed to the @+RTS -N@ flag. This is the number of
229 -- Haskell threads that can run truly simultaneously at any given
230 -- time, and is typically set to the number of physical CPU cores on
232 numCapabilities :: Int
233 numCapabilities = unsafePerformIO $ do
234 n <- peek n_capabilities
235 return (fromIntegral n)
237 foreign import ccall "&n_capabilities" n_capabilities :: Ptr CInt
239 childHandler :: SomeException -> IO ()
240 childHandler err = catchException (real_handler err) childHandler
242 real_handler :: SomeException -> IO ()
243 real_handler se@(SomeException ex) =
244 -- ignore thread GC and killThread exceptions:
246 Just BlockedOnDeadMVar -> return ()
248 Just BlockedIndefinitely -> return ()
250 Just ThreadKilled -> return ()
252 -- report all others:
253 Just StackOverflow -> reportStackOverflow
256 {- | 'killThread' terminates the given thread (GHC only).
257 Any work already done by the thread isn\'t
258 lost: the computation is suspended until required by another thread.
259 The memory used by the thread will be garbage collected if it isn\'t
260 referenced from anywhere. The 'killThread' function is defined in
263 > killThread tid = throwTo tid (AsyncException ThreadKilled)
266 killThread :: ThreadId -> IO ()
267 killThread tid = throwTo tid (toException ThreadKilled)
269 {- | 'throwTo' raises an arbitrary exception in the target thread (GHC only).
271 'throwTo' does not return until the exception has been raised in the
273 The calling thread can thus be certain that the target
274 thread has received the exception. This is a useful property to know
275 when dealing with race conditions: eg. if there are two threads that
276 can kill each other, it is guaranteed that only one of the threads
277 will get to kill the other.
279 If the target thread is currently making a foreign call, then the
280 exception will not be raised (and hence 'throwTo' will not return)
281 until the call has completed. This is the case regardless of whether
282 the call is inside a 'block' or not.
284 Important note: the behaviour of 'throwTo' differs from that described in
285 the paper \"Asynchronous exceptions in Haskell\"
286 (<http://research.microsoft.com/~simonpj/Papers/asynch-exns.htm>).
287 In the paper, 'throwTo' is non-blocking; but the library implementation adopts
288 a more synchronous design in which 'throwTo' does not return until the exception
289 is received by the target thread. The trade-off is discussed in Section 8 of the paper.
290 Like any blocking operation, 'throwTo' is therefore interruptible (see Section 4.3 of
293 There is currently no guarantee that the exception delivered by 'throwTo' will be
294 delivered at the first possible opportunity. In particular, if a thread may
295 unblock and then re-block exceptions (using 'unblock' and 'block') without receiving
296 a pending 'throwTo'. This is arguably undesirable behaviour.
299 -- XXX This is duplicated in Control.{Old,}Exception
300 throwTo :: ThreadId -> SomeException -> IO ()
301 throwTo (ThreadId id) ex = IO $ \ s ->
302 case (killThread# id ex s) of s1 -> (# s1, () #)
304 -- | Returns the 'ThreadId' of the calling thread (GHC only).
305 myThreadId :: IO ThreadId
306 myThreadId = IO $ \s ->
307 case (myThreadId# s) of (# s1, id #) -> (# s1, ThreadId id #)
310 -- |The 'yield' action allows (forces, in a co-operative multitasking
311 -- implementation) a context-switch to any other currently runnable
312 -- threads (if any), and is occasionally useful when implementing
313 -- concurrency abstractions.
316 case (yield# s) of s1 -> (# s1, () #)
318 {- | 'labelThread' stores a string as identifier for this thread if
319 you built a RTS with debugging support. This identifier will be used in
320 the debugging output to make distinction of different threads easier
321 (otherwise you only have the thread state object\'s address in the heap).
323 Other applications like the graphical Concurrent Haskell Debugger
324 (<http://www.informatik.uni-kiel.de/~fhu/chd/>) may choose to overload
325 'labelThread' for their purposes as well.
328 labelThread :: ThreadId -> String -> IO ()
329 labelThread (ThreadId t) str = IO $ \ s ->
330 let ps = packCString# str
331 adr = byteArrayContents# ps in
332 case (labelThread# t adr s) of s1 -> (# s1, () #)
334 -- Nota Bene: 'pseq' used to be 'seq'
335 -- but 'seq' is now defined in PrelGHC
337 -- "pseq" is defined a bit weirdly (see below)
339 -- The reason for the strange "lazy" call is that
340 -- it fools the compiler into thinking that pseq and par are non-strict in
341 -- their second argument (even if it inlines pseq at the call site).
342 -- If it thinks pseq is strict in "y", then it often evaluates
343 -- "y" before "x", which is totally wrong.
347 pseq x y = x `seq` lazy y
351 par x y = case (par# x) of { _ -> lazy y }
356 -- ^blocked on on 'MVar'
358 -- ^blocked on a computation in progress by another thread
360 -- ^blocked in 'throwTo'
362 -- ^blocked in 'retry' in an STM transaction
363 | BlockedOnForeignCall
364 -- ^currently in a foreign call
366 -- ^blocked on some other resource. Without @-threaded@,
367 -- I\/O and 'threadDelay' show up as 'BlockedOnOther', with @-threaded@
368 -- they show up as 'BlockedOnMVar'.
369 deriving (Eq,Ord,Show)
371 -- | The current status of a thread
374 -- ^the thread is currently runnable or running
376 -- ^the thread has finished
377 | ThreadBlocked BlockReason
378 -- ^the thread is blocked on some resource
380 -- ^the thread received an uncaught exception
381 deriving (Eq,Ord,Show)
383 threadStatus :: ThreadId -> IO ThreadStatus
384 threadStatus (ThreadId t) = IO $ \s ->
385 case threadStatus# t s of
386 (# s', stat #) -> (# s', mk_stat (I# stat) #)
388 -- NB. keep these in sync with includes/Constants.h
389 mk_stat 0 = ThreadRunning
390 mk_stat 1 = ThreadBlocked BlockedOnMVar
391 mk_stat 2 = ThreadBlocked BlockedOnBlackHole
392 mk_stat 3 = ThreadBlocked BlockedOnException
393 mk_stat 7 = ThreadBlocked BlockedOnSTM
394 mk_stat 11 = ThreadBlocked BlockedOnForeignCall
395 mk_stat 12 = ThreadBlocked BlockedOnForeignCall
396 mk_stat 16 = ThreadFinished
397 mk_stat 17 = ThreadDied
398 mk_stat _ = ThreadBlocked BlockedOnOther
402 %************************************************************************
404 \subsection[stm]{Transactional heap operations}
406 %************************************************************************
408 TVars are shared memory locations which support atomic memory
412 -- |A monad supporting atomic memory transactions.
413 newtype STM a = STM (State# RealWorld -> (# State# RealWorld, a #))
415 unSTM :: STM a -> (State# RealWorld -> (# State# RealWorld, a #))
418 INSTANCE_TYPEABLE1(STM,stmTc,"STM")
420 instance Functor STM where
421 fmap f x = x >>= (return . f)
423 instance Monad STM where
424 {-# INLINE return #-}
428 return x = returnSTM x
429 m >>= k = bindSTM m k
431 bindSTM :: STM a -> (a -> STM b) -> STM b
432 bindSTM (STM m) k = STM ( \s ->
434 (# new_s, a #) -> unSTM (k a) new_s
437 thenSTM :: STM a -> STM b -> STM b
438 thenSTM (STM m) k = STM ( \s ->
440 (# new_s, a #) -> unSTM k new_s
443 returnSTM :: a -> STM a
444 returnSTM x = STM (\s -> (# s, x #))
446 -- | Unsafely performs IO in the STM monad. Beware: this is a highly
447 -- dangerous thing to do.
449 -- * The STM implementation will often run transactions multiple
450 -- times, so you need to be prepared for this if your IO has any
453 -- * The STM implementation will abort transactions that are known to
454 -- be invalid and need to be restarted. This may happen in the middle
455 -- of `unsafeIOToSTM`, so make sure you don't acquire any resources
456 -- that need releasing (exception handlers are ignored when aborting
457 -- the transaction). That includes doing any IO using Handles, for
458 -- example. Getting this wrong will probably lead to random deadlocks.
460 -- * The transaction may have seen an inconsistent view of memory when
461 -- the IO runs. Invariants that you expect to be true throughout
462 -- your program may not be true inside a transaction, due to the
463 -- way transactions are implemented. Normally this wouldn't be visible
464 -- to the programmer, but using `unsafeIOToSTM` can expose it.
466 unsafeIOToSTM :: IO a -> STM a
467 unsafeIOToSTM (IO m) = STM m
469 -- |Perform a series of STM actions atomically.
471 -- You cannot use 'atomically' inside an 'unsafePerformIO' or 'unsafeInterleaveIO'.
472 -- Any attempt to do so will result in a runtime error. (Reason: allowing
473 -- this would effectively allow a transaction inside a transaction, depending
474 -- on exactly when the thunk is evaluated.)
476 -- However, see 'newTVarIO', which can be called inside 'unsafePerformIO',
477 -- and which allows top-level TVars to be allocated.
479 atomically :: STM a -> IO a
480 atomically (STM m) = IO (\s -> (atomically# m) s )
482 -- |Retry execution of the current memory transaction because it has seen
483 -- values in TVars which mean that it should not continue (e.g. the TVars
484 -- represent a shared buffer that is now empty). The implementation may
485 -- block the thread until one of the TVars that it has read from has been
486 -- udpated. (GHC only)
488 retry = STM $ \s# -> retry# s#
490 -- |Compose two alternative STM actions (GHC only). If the first action
491 -- completes without retrying then it forms the result of the orElse.
492 -- Otherwise, if the first action retries, then the second action is
493 -- tried in its place. If both actions retry then the orElse as a
495 orElse :: STM a -> STM a -> STM a
496 orElse (STM m) e = STM $ \s -> catchRetry# m (unSTM e) s
498 -- |Exception handling within STM actions.
499 catchSTM :: STM a -> (SomeException -> STM a) -> STM a
500 catchSTM (STM m) k = STM $ \s -> catchSTM# m (\ex -> unSTM (k ex)) s
502 -- | Low-level primitive on which always and alwaysSucceeds are built.
503 -- checkInv differs form these in that (i) the invariant is not
504 -- checked when checkInv is called, only at the end of this and
505 -- subsequent transcations, (ii) the invariant failure is indicated
506 -- by raising an exception.
507 checkInv :: STM a -> STM ()
508 checkInv (STM m) = STM (\s -> (check# m) s)
510 -- | alwaysSucceeds adds a new invariant that must be true when passed
511 -- to alwaysSucceeds, at the end of the current transaction, and at
512 -- the end of every subsequent transaction. If it fails at any
513 -- of those points then the transaction violating it is aborted
514 -- and the exception raised by the invariant is propagated.
515 alwaysSucceeds :: STM a -> STM ()
516 alwaysSucceeds i = do ( do i ; retry ) `orElse` ( return () )
519 -- | always is a variant of alwaysSucceeds in which the invariant is
520 -- expressed as an STM Bool action that must return True. Returning
521 -- False or raising an exception are both treated as invariant failures.
522 always :: STM Bool -> STM ()
523 always i = alwaysSucceeds ( do v <- i
524 if (v) then return () else ( error "Transacional invariant violation" ) )
526 -- |Shared memory locations that support atomic memory transactions.
527 data TVar a = TVar (TVar# RealWorld a)
529 INSTANCE_TYPEABLE1(TVar,tvarTc,"TVar")
531 instance Eq (TVar a) where
532 (TVar tvar1#) == (TVar tvar2#) = sameTVar# tvar1# tvar2#
534 -- |Create a new TVar holding a value supplied
535 newTVar :: a -> STM (TVar a)
536 newTVar val = STM $ \s1# ->
537 case newTVar# val s1# of
538 (# s2#, tvar# #) -> (# s2#, TVar tvar# #)
540 -- |@IO@ version of 'newTVar'. This is useful for creating top-level
541 -- 'TVar's using 'System.IO.Unsafe.unsafePerformIO', because using
542 -- 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' isn't
544 newTVarIO :: a -> IO (TVar a)
545 newTVarIO val = IO $ \s1# ->
546 case newTVar# val s1# of
547 (# s2#, tvar# #) -> (# s2#, TVar tvar# #)
549 -- |Return the current value stored in a TVar
550 readTVar :: TVar a -> STM a
551 readTVar (TVar tvar#) = STM $ \s# -> readTVar# tvar# s#
553 -- |Write the supplied value into a TVar
554 writeTVar :: TVar a -> a -> STM ()
555 writeTVar (TVar tvar#) val = STM $ \s1# ->
556 case writeTVar# tvar# val s1# of
561 %************************************************************************
563 \subsection[mvars]{M-Structures}
565 %************************************************************************
567 M-Vars are rendezvous points for concurrent threads. They begin
568 empty, and any attempt to read an empty M-Var blocks. When an M-Var
569 is written, a single blocked thread may be freed. Reading an M-Var
570 toggles its state from full back to empty. Therefore, any value
571 written to an M-Var may only be read once. Multiple reads and writes
572 are allowed, but there must be at least one read between any two
576 --Defined in IOBase to avoid cycle: data MVar a = MVar (SynchVar# RealWorld a)
578 -- |Create an 'MVar' which is initially empty.
579 newEmptyMVar :: IO (MVar a)
580 newEmptyMVar = IO $ \ s# ->
582 (# s2#, svar# #) -> (# s2#, MVar svar# #)
584 -- |Create an 'MVar' which contains the supplied value.
585 newMVar :: a -> IO (MVar a)
587 newEmptyMVar >>= \ mvar ->
588 putMVar mvar value >>
591 -- |Return the contents of the 'MVar'. If the 'MVar' is currently
592 -- empty, 'takeMVar' will wait until it is full. After a 'takeMVar',
593 -- the 'MVar' is left empty.
595 -- There are two further important properties of 'takeMVar':
597 -- * 'takeMVar' is single-wakeup. That is, if there are multiple
598 -- threads blocked in 'takeMVar', and the 'MVar' becomes full,
599 -- only one thread will be woken up. The runtime guarantees that
600 -- the woken thread completes its 'takeMVar' operation.
602 -- * When multiple threads are blocked on an 'MVar', they are
603 -- woken up in FIFO order. This is useful for providing
604 -- fairness properties of abstractions built using 'MVar's.
606 takeMVar :: MVar a -> IO a
607 takeMVar (MVar mvar#) = IO $ \ s# -> takeMVar# mvar# s#
609 -- |Put a value into an 'MVar'. If the 'MVar' is currently full,
610 -- 'putMVar' will wait until it becomes empty.
612 -- There are two further important properties of 'putMVar':
614 -- * 'putMVar' is single-wakeup. That is, if there are multiple
615 -- threads blocked in 'putMVar', and the 'MVar' becomes empty,
616 -- only one thread will be woken up. The runtime guarantees that
617 -- the woken thread completes its 'putMVar' operation.
619 -- * When multiple threads are blocked on an 'MVar', they are
620 -- woken up in FIFO order. This is useful for providing
621 -- fairness properties of abstractions built using 'MVar's.
623 putMVar :: MVar a -> a -> IO ()
624 putMVar (MVar mvar#) x = IO $ \ s# ->
625 case putMVar# mvar# x s# of
628 -- |A non-blocking version of 'takeMVar'. The 'tryTakeMVar' function
629 -- returns immediately, with 'Nothing' if the 'MVar' was empty, or
630 -- @'Just' a@ if the 'MVar' was full with contents @a@. After 'tryTakeMVar',
631 -- the 'MVar' is left empty.
632 tryTakeMVar :: MVar a -> IO (Maybe a)
633 tryTakeMVar (MVar m) = IO $ \ s ->
634 case tryTakeMVar# m s of
635 (# s, 0#, _ #) -> (# s, Nothing #) -- MVar is empty
636 (# s, _, a #) -> (# s, Just a #) -- MVar is full
638 -- |A non-blocking version of 'putMVar'. The 'tryPutMVar' function
639 -- attempts to put the value @a@ into the 'MVar', returning 'True' if
640 -- it was successful, or 'False' otherwise.
641 tryPutMVar :: MVar a -> a -> IO Bool
642 tryPutMVar (MVar mvar#) x = IO $ \ s# ->
643 case tryPutMVar# mvar# x s# of
644 (# s, 0# #) -> (# s, False #)
645 (# s, _ #) -> (# s, True #)
647 -- |Check whether a given 'MVar' is empty.
649 -- Notice that the boolean value returned is just a snapshot of
650 -- the state of the MVar. By the time you get to react on its result,
651 -- the MVar may have been filled (or emptied) - so be extremely
652 -- careful when using this operation. Use 'tryTakeMVar' instead if possible.
653 isEmptyMVar :: MVar a -> IO Bool
654 isEmptyMVar (MVar mv#) = IO $ \ s# ->
655 case isEmptyMVar# mv# s# of
656 (# s2#, flg #) -> (# s2#, not (flg ==# 0#) #)
658 -- |Add a finalizer to an 'MVar' (GHC only). See "Foreign.ForeignPtr" and
659 -- "System.Mem.Weak" for more about finalizers.
660 addMVarFinalizer :: MVar a -> IO () -> IO ()
661 addMVarFinalizer (MVar m) finalizer =
662 IO $ \s -> case mkWeak# m () finalizer s of { (# s1, w #) -> (# s1, () #) }
664 withMVar :: MVar a -> (a -> IO b) -> IO b
668 b <- catchAny (unblock (io a))
669 (\e -> do putMVar m a; throw e)
675 %************************************************************************
677 \subsection{Thread waiting}
679 %************************************************************************
682 #ifdef mingw32_HOST_OS
684 -- Note: threadWaitRead and threadWaitWrite aren't really functional
685 -- on Win32, but left in there because lib code (still) uses them (the manner
686 -- in which they're used doesn't cause problems on a Win32 platform though.)
688 asyncRead :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
689 asyncRead (I# fd) (I# isSock) (I# len) (Ptr buf) =
690 IO $ \s -> case asyncRead# fd isSock len buf s of
691 (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
693 asyncWrite :: Int -> Int -> Int -> Ptr a -> IO (Int, Int)
694 asyncWrite (I# fd) (I# isSock) (I# len) (Ptr buf) =
695 IO $ \s -> case asyncWrite# fd isSock len buf s of
696 (# s, len#, err# #) -> (# s, (I# len#, I# err#) #)
698 asyncDoProc :: FunPtr (Ptr a -> IO Int) -> Ptr a -> IO Int
699 asyncDoProc (FunPtr proc) (Ptr param) =
700 -- the 'length' value is ignored; simplifies implementation of
701 -- the async*# primops to have them all return the same result.
702 IO $ \s -> case asyncDoProc# proc param s of
703 (# s, len#, err# #) -> (# s, I# err# #)
705 -- to aid the use of these primops by the IO Handle implementation,
706 -- provide the following convenience funs:
708 -- this better be a pinned byte array!
709 asyncReadBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
710 asyncReadBA fd isSock len off bufB =
711 asyncRead fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
713 asyncWriteBA :: Int -> Int -> Int -> Int -> MutableByteArray# RealWorld -> IO (Int,Int)
714 asyncWriteBA fd isSock len off bufB =
715 asyncWrite fd isSock len ((Ptr (byteArrayContents# (unsafeCoerce# bufB))) `plusPtr` off)
719 -- -----------------------------------------------------------------------------
722 -- | Block the current thread until data is available to read on the
723 -- given file descriptor (GHC only).
724 threadWaitRead :: Fd -> IO ()
726 #ifndef mingw32_HOST_OS
727 | threaded = waitForReadEvent fd
729 | otherwise = IO $ \s ->
730 case fromIntegral fd of { I# fd# ->
731 case waitRead# fd# s of { s -> (# s, () #)
734 -- | Block the current thread until data can be written to the
735 -- given file descriptor (GHC only).
736 threadWaitWrite :: Fd -> IO ()
738 #ifndef mingw32_HOST_OS
739 | threaded = waitForWriteEvent fd
741 | otherwise = IO $ \s ->
742 case fromIntegral fd of { I# fd# ->
743 case waitWrite# fd# s of { s -> (# s, () #)
746 -- | Suspends the current thread for a given number of microseconds
749 -- There is no guarantee that the thread will be rescheduled promptly
750 -- when the delay has expired, but the thread will never continue to
751 -- run /earlier/ than specified.
753 threadDelay :: Int -> IO ()
755 | threaded = waitForDelayEvent time
756 | otherwise = IO $ \s ->
757 case fromIntegral time of { I# time# ->
758 case delay# time# s of { s -> (# s, () #)
762 -- | Set the value of returned TVar to True after a given number of
763 -- microseconds. The caveats associated with threadDelay also apply.
765 registerDelay :: Int -> IO (TVar Bool)
767 | threaded = waitForDelayEventSTM usecs
768 | otherwise = error "registerDelay: requires -threaded"
770 foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
772 waitForDelayEvent :: Int -> IO ()
773 waitForDelayEvent usecs = do
775 target <- calculateTarget usecs
776 atomicModifyIORef pendingDelays (\xs -> (Delay target m : xs, ()))
780 -- Delays for use in STM
781 waitForDelayEventSTM :: Int -> IO (TVar Bool)
782 waitForDelayEventSTM usecs = do
783 t <- atomically $ newTVar False
784 target <- calculateTarget usecs
785 atomicModifyIORef pendingDelays (\xs -> (DelaySTM target t : xs, ()))
789 calculateTarget :: Int -> IO USecs
790 calculateTarget usecs = do
792 return $ now + (fromIntegral usecs)
795 -- ----------------------------------------------------------------------------
796 -- Threaded RTS implementation of threadWaitRead, threadWaitWrite, threadDelay
798 -- In the threaded RTS, we employ a single IO Manager thread to wait
799 -- for all outstanding IO requests (threadWaitRead,threadWaitWrite)
800 -- and delays (threadDelay).
802 -- We can do this because in the threaded RTS the IO Manager can make
803 -- a non-blocking call to select(), so we don't have to do select() in
804 -- the scheduler as we have to in the non-threaded RTS. We get performance
805 -- benefits from doing it this way, because we only have to restart the select()
806 -- when a new request arrives, rather than doing one select() each time
807 -- around the scheduler loop. Furthermore, the scheduler can be simplified
808 -- by not having to check for completed IO requests.
810 -- Issues, possible problems:
812 -- - we might want bound threads to just do the blocking
813 -- operation rather than communicating with the IO manager
814 -- thread. This would prevent simgle-threaded programs which do
815 -- IO from requiring multiple OS threads. However, it would also
816 -- prevent bound threads waiting on IO from being killed or sent
819 -- - Apprently exec() doesn't work on Linux in a multithreaded program.
820 -- I couldn't repeat this.
822 -- - How do we handle signal delivery in the multithreaded RTS?
824 -- - forkProcess will kill the IO manager thread. Let's just
825 -- hope we don't need to do any blocking IO between fork & exec.
827 #ifndef mingw32_HOST_OS
829 = Read {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
830 | Write {-# UNPACK #-} !Fd {-# UNPACK #-} !(MVar ())
834 = Delay {-# UNPACK #-} !USecs {-# UNPACK #-} !(MVar ())
835 | DelaySTM {-# UNPACK #-} !USecs {-# UNPACK #-} !(TVar Bool)
837 #ifndef mingw32_HOST_OS
838 pendingEvents :: IORef [IOReq]
840 pendingDelays :: IORef [DelayReq]
841 -- could use a strict list or array here
842 {-# NOINLINE pendingEvents #-}
843 {-# NOINLINE pendingDelays #-}
844 (pendingEvents,pendingDelays) = unsafePerformIO $ do
849 -- the first time we schedule an IO request, the service thread
850 -- will be created (cool, huh?)
852 ensureIOManagerIsRunning :: IO ()
853 ensureIOManagerIsRunning
854 | threaded = seq pendingEvents $ return ()
855 | otherwise = return ()
857 insertDelay :: DelayReq -> [DelayReq] -> [DelayReq]
858 insertDelay d [] = [d]
859 insertDelay d1 ds@(d2 : rest)
860 | delayTime d1 <= delayTime d2 = d1 : ds
861 | otherwise = d2 : insertDelay d1 rest
863 delayTime :: DelayReq -> USecs
864 delayTime (Delay t _) = t
865 delayTime (DelaySTM t _) = t
869 -- XXX: move into GHC.IOBase from Data.IORef?
870 atomicModifyIORef :: IORef a -> (a -> (a,b)) -> IO b
871 atomicModifyIORef (IORef (STRef r#)) f = IO $ \s -> atomicModifyMutVar# r# f s
873 foreign import ccall unsafe "getUSecOfDay"
874 getUSecOfDay :: IO USecs
876 prodding :: IORef Bool
877 {-# NOINLINE prodding #-}
878 prodding = unsafePerformIO (newIORef False)
880 prodServiceThread :: IO ()
881 prodServiceThread = do
882 was_set <- atomicModifyIORef prodding (\a -> (True,a))
883 if (not (was_set)) then wakeupIOManager else return ()
885 #ifdef mingw32_HOST_OS
886 -- ----------------------------------------------------------------------------
887 -- Windows IO manager thread
889 startIOManagerThread :: IO ()
890 startIOManagerThread = do
891 wakeup <- c_getIOManagerEvent
892 forkIO $ service_loop wakeup []
895 service_loop :: HANDLE -- read end of pipe
896 -> [DelayReq] -- current delay requests
899 service_loop wakeup old_delays = do
900 -- pick up new delay requests
901 new_delays <- atomicModifyIORef pendingDelays (\a -> ([],a))
902 let delays = foldr insertDelay old_delays new_delays
905 (delays', timeout) <- getDelay now delays
907 r <- c_WaitForSingleObject wakeup timeout
909 0xffffffff -> do c_maperrno; throwErrno "service_loop"
911 r <- c_readIOManagerEvent
914 _ | r == io_MANAGER_WAKEUP -> return False
915 _ | r == io_MANAGER_DIE -> return True
916 0 -> return False -- spurious wakeup
917 r -> do start_console_handler (r `shiftR` 1); return False
920 else service_cont wakeup delays'
922 _other -> service_cont wakeup delays' -- probably timeout
924 service_cont wakeup delays = do
925 atomicModifyIORef prodding (\_ -> (False,False))
926 service_loop wakeup delays
928 -- must agree with rts/win32/ThrIOManager.c
929 io_MANAGER_WAKEUP = 0xffffffff :: Word32
930 io_MANAGER_DIE = 0xfffffffe :: Word32
936 -- these are sent to Services only.
939 deriving (Eq, Ord, Enum, Show, Read, Typeable)
941 start_console_handler :: Word32 -> IO ()
942 start_console_handler r =
943 case toWin32ConsoleEvent r of
944 Just x -> withMVar win32ConsoleHandler $ \handler -> do
949 toWin32ConsoleEvent ev =
951 0 {- CTRL_C_EVENT-} -> Just ControlC
952 1 {- CTRL_BREAK_EVENT-} -> Just Break
953 2 {- CTRL_CLOSE_EVENT-} -> Just Close
954 5 {- CTRL_LOGOFF_EVENT-} -> Just Logoff
955 6 {- CTRL_SHUTDOWN_EVENT-} -> Just Shutdown
958 win32ConsoleHandler :: MVar (ConsoleEvent -> IO ())
959 win32ConsoleHandler = unsafePerformIO (newMVar (error "win32ConsoleHandler"))
961 stick :: IORef HANDLE
962 {-# NOINLINE stick #-}
963 stick = unsafePerformIO (newIORef nullPtr)
966 hdl <- readIORef stick
967 c_sendIOManagerEvent io_MANAGER_WAKEUP
969 -- Walk the queue of pending delays, waking up any that have passed
970 -- and return the smallest delay to wait for. The queue of pending
971 -- delays is kept ordered.
972 getDelay :: USecs -> [DelayReq] -> IO ([DelayReq], DWORD)
973 getDelay now [] = return ([], iNFINITE)
974 getDelay now all@(d : rest)
976 Delay time m | now >= time -> do
979 DelaySTM time t | now >= time -> do
980 atomically $ writeTVar t True
983 -- delay is in millisecs for WaitForSingleObject
984 let micro_seconds = delayTime d - now
985 milli_seconds = (micro_seconds + 999) `div` 1000
986 in return (all, fromIntegral milli_seconds)
988 -- ToDo: this just duplicates part of System.Win32.Types, which isn't
989 -- available yet. We should move some Win32 functionality down here,
990 -- maybe as part of the grand reorganisation of the base package...
994 iNFINITE = 0xFFFFFFFF :: DWORD -- urgh
996 foreign import ccall unsafe "getIOManagerEvent" -- in the RTS (ThrIOManager.c)
997 c_getIOManagerEvent :: IO HANDLE
999 foreign import ccall unsafe "readIOManagerEvent" -- in the RTS (ThrIOManager.c)
1000 c_readIOManagerEvent :: IO Word32
1002 foreign import ccall unsafe "sendIOManagerEvent" -- in the RTS (ThrIOManager.c)
1003 c_sendIOManagerEvent :: Word32 -> IO ()
1005 foreign import ccall unsafe "maperrno" -- in Win32Utils.c
1008 foreign import stdcall "WaitForSingleObject"
1009 c_WaitForSingleObject :: HANDLE -> DWORD -> IO DWORD
1012 -- ----------------------------------------------------------------------------
1013 -- Unix IO manager thread, using select()
1015 startIOManagerThread :: IO ()
1016 startIOManagerThread = do
1017 allocaArray 2 $ \fds -> do
1018 throwErrnoIfMinus1 "startIOManagerThread" (c_pipe fds)
1019 rd_end <- peekElemOff fds 0
1020 wr_end <- peekElemOff fds 1
1021 writeIORef stick (fromIntegral wr_end)
1022 c_setIOManagerPipe wr_end
1024 allocaBytes sizeofFdSet $ \readfds -> do
1025 allocaBytes sizeofFdSet $ \writefds -> do
1026 allocaBytes sizeofTimeVal $ \timeval -> do
1027 service_loop (fromIntegral rd_end) readfds writefds timeval [] []
1031 :: Fd -- listen to this for wakeup calls
1038 service_loop wakeup readfds writefds ptimeval old_reqs old_delays = do
1040 -- pick up new IO requests
1041 new_reqs <- atomicModifyIORef pendingEvents (\a -> ([],a))
1042 let reqs = new_reqs ++ old_reqs
1044 -- pick up new delay requests
1045 new_delays <- atomicModifyIORef pendingDelays (\a -> ([],a))
1046 let delays = foldr insertDelay old_delays new_delays
1048 -- build the FDSets for select()
1051 fdSet wakeup readfds
1052 maxfd <- buildFdSets 0 readfds writefds reqs
1054 -- perform the select()
1055 let do_select delays = do
1056 -- check the current time and wake up any thread in
1057 -- threadDelay whose timeout has expired. Also find the
1058 -- timeout value for the select() call.
1060 (delays', timeout) <- getDelay now ptimeval delays
1062 res <- c_select (fromIntegral ((max wakeup maxfd)+1)) readfds writefds
1068 _ | err == eINTR -> do_select delays'
1069 -- EINTR: just redo the select()
1070 _ | err == eBADF -> return (True, delays)
1071 -- EBADF: one of the file descriptors is closed or bad,
1072 -- we don't know which one, so wake everyone up.
1073 _ | otherwise -> throwErrno "select"
1074 -- otherwise (ENOMEM or EINVAL) something has gone
1075 -- wrong; report the error.
1077 return (False,delays')
1079 (wakeup_all,delays') <- do_select delays
1082 if wakeup_all then return False
1084 b <- fdIsSet wakeup readfds
1087 else alloca $ \p -> do
1088 c_read (fromIntegral wakeup) p 1; return ()
1091 _ | s == io_MANAGER_WAKEUP -> return False
1092 _ | s == io_MANAGER_DIE -> return True
1093 _ -> withMVar signalHandlerLock $ \_ -> do
1094 handler_tbl <- peek handlers
1095 sp <- peekElemOff handler_tbl (fromIntegral s)
1096 io <- deRefStablePtr sp
1100 if exit then return () else do
1102 atomicModifyIORef prodding (\_ -> (False,False))
1104 reqs' <- if wakeup_all then do wakeupAll reqs; return []
1105 else completeRequests reqs readfds writefds []
1107 service_loop wakeup readfds writefds ptimeval reqs' delays'
1109 io_MANAGER_WAKEUP = 0xff :: CChar
1110 io_MANAGER_DIE = 0xfe :: CChar
1113 {-# NOINLINE stick #-}
1114 stick = unsafePerformIO (newIORef 0)
1116 wakeupIOManager :: IO ()
1117 wakeupIOManager = do
1118 fd <- readIORef stick
1119 with io_MANAGER_WAKEUP $ \pbuf -> do
1120 c_write (fromIntegral fd) pbuf 1; return ()
1122 -- Lock used to protect concurrent access to signal_handlers. Symptom of
1123 -- this race condition is #1922, although that bug was on Windows a similar
1124 -- bug also exists on Unix.
1125 signalHandlerLock :: MVar ()
1126 signalHandlerLock = unsafePerformIO (newMVar ())
1128 foreign import ccall "&signal_handlers" handlers :: Ptr (Ptr (StablePtr (IO ())))
1130 foreign import ccall "setIOManagerPipe"
1131 c_setIOManagerPipe :: CInt -> IO ()
1133 -- -----------------------------------------------------------------------------
1136 buildFdSets maxfd readfds writefds [] = return maxfd
1137 buildFdSets maxfd readfds writefds (Read fd m : reqs)
1138 | fd >= fD_SETSIZE = error "buildFdSets: file descriptor out of range"
1141 buildFdSets (max maxfd fd) readfds writefds reqs
1142 buildFdSets maxfd readfds writefds (Write fd m : reqs)
1143 | fd >= fD_SETSIZE = error "buildFdSets: file descriptor out of range"
1146 buildFdSets (max maxfd fd) readfds writefds reqs
1148 completeRequests [] _ _ reqs' = return reqs'
1149 completeRequests (Read fd m : reqs) readfds writefds reqs' = do
1150 b <- fdIsSet fd readfds
1152 then do putMVar m (); completeRequests reqs readfds writefds reqs'
1153 else completeRequests reqs readfds writefds (Read fd m : reqs')
1154 completeRequests (Write fd m : reqs) readfds writefds reqs' = do
1155 b <- fdIsSet fd writefds
1157 then do putMVar m (); completeRequests reqs readfds writefds reqs'
1158 else completeRequests reqs readfds writefds (Write fd m : reqs')
1160 wakeupAll [] = return ()
1161 wakeupAll (Read fd m : reqs) = do putMVar m (); wakeupAll reqs
1162 wakeupAll (Write fd m : reqs) = do putMVar m (); wakeupAll reqs
1164 waitForReadEvent :: Fd -> IO ()
1165 waitForReadEvent fd = do
1167 atomicModifyIORef pendingEvents (\xs -> (Read fd m : xs, ()))
1171 waitForWriteEvent :: Fd -> IO ()
1172 waitForWriteEvent fd = do
1174 atomicModifyIORef pendingEvents (\xs -> (Write fd m : xs, ()))
1178 -- -----------------------------------------------------------------------------
1181 -- Walk the queue of pending delays, waking up any that have passed
1182 -- and return the smallest delay to wait for. The queue of pending
1183 -- delays is kept ordered.
1184 getDelay :: USecs -> Ptr CTimeVal -> [DelayReq] -> IO ([DelayReq], Ptr CTimeVal)
1185 getDelay now ptimeval [] = return ([],nullPtr)
1186 getDelay now ptimeval all@(d : rest)
1188 Delay time m | now >= time -> do
1190 getDelay now ptimeval rest
1191 DelaySTM time t | now >= time -> do
1192 atomically $ writeTVar t True
1193 getDelay now ptimeval rest
1195 setTimevalTicks ptimeval (delayTime d - now)
1196 return (all,ptimeval)
1198 newtype CTimeVal = CTimeVal ()
1200 foreign import ccall unsafe "sizeofTimeVal"
1201 sizeofTimeVal :: Int
1203 foreign import ccall unsafe "setTimevalTicks"
1204 setTimevalTicks :: Ptr CTimeVal -> USecs -> IO ()
1207 On Win32 we're going to have a single Pipe, and a
1208 waitForSingleObject with the delay time. For signals, we send a
1209 byte down the pipe just like on Unix.
1212 -- ----------------------------------------------------------------------------
1213 -- select() interface
1215 -- ToDo: move to System.Posix.Internals?
1217 newtype CFdSet = CFdSet ()
1219 foreign import ccall safe "select"
1220 c_select :: CInt -> Ptr CFdSet -> Ptr CFdSet -> Ptr CFdSet -> Ptr CTimeVal
1223 foreign import ccall unsafe "hsFD_SETSIZE"
1224 c_fD_SETSIZE :: CInt
1227 fD_SETSIZE = fromIntegral c_fD_SETSIZE
1229 foreign import ccall unsafe "hsFD_CLR"
1230 c_fdClr :: CInt -> Ptr CFdSet -> IO ()
1232 fdClr :: Fd -> Ptr CFdSet -> IO ()
1233 fdClr (Fd fd) fdset = c_fdClr fd fdset
1235 foreign import ccall unsafe "hsFD_ISSET"
1236 c_fdIsSet :: CInt -> Ptr CFdSet -> IO CInt
1238 fdIsSet :: Fd -> Ptr CFdSet -> IO CInt
1239 fdIsSet (Fd fd) fdset = c_fdIsSet fd fdset
1241 foreign import ccall unsafe "hsFD_SET"
1242 c_fdSet :: CInt -> Ptr CFdSet -> IO ()
1244 fdSet :: Fd -> Ptr CFdSet -> IO ()
1245 fdSet (Fd fd) fdset = c_fdSet fd fdset
1247 foreign import ccall unsafe "hsFD_ZERO"
1248 fdZero :: Ptr CFdSet -> IO ()
1250 foreign import ccall unsafe "sizeof_fd_set"