1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
{-# LANGUAGE Unsafe #-}
{-# LANGUAGE BangPatterns, ForeignFunctionInterface, NoImplicitPrelude #-}
module GHC.Event.Thread
( getSystemEventManager
, ensureIOManagerIsRunning
, threadWaitRead
, threadWaitWrite
, closeFdWith
, threadDelay
, registerDelay
) where
import Data.IORef (IORef, newIORef, readIORef, writeIORef)
import Data.Maybe (Maybe(..))
import Foreign.C.Error (eBADF, errnoToIOError)
import Foreign.Ptr (Ptr)
import GHC.Base
import GHC.Conc.Sync (TVar, ThreadId, ThreadStatus(..), atomically, forkIO,
labelThread, modifyMVar_, newTVar, sharedCAF,
threadStatus, writeTVar)
import GHC.IO (mask_, onException)
import GHC.IO.Exception (ioError)
import GHC.MVar (MVar, newEmptyMVar, newMVar, putMVar, takeMVar)
import GHC.Event.Internal (eventIs, evtClose)
import GHC.Event.Manager (Event, EventManager, evtRead, evtWrite, loop,
new, registerFd, unregisterFd_, registerTimeout)
import qualified GHC.Event.Manager as M
import System.IO.Unsafe (unsafePerformIO)
import System.Posix.Types (Fd)
-- | Suspends the current thread for a given number of microseconds
-- (GHC only).
--
-- There is no guarantee that the thread will be rescheduled promptly
-- when the delay has expired, but the thread will never continue to
-- run /earlier/ than specified.
threadDelay :: Int -> IO ()
threadDelay usecs = mask_ $ do
Just mgr <- getSystemEventManager
m <- newEmptyMVar
reg <- registerTimeout mgr usecs (putMVar m ())
takeMVar m `onException` M.unregisterTimeout mgr reg
-- | Set the value of returned TVar to True after a given number of
-- microseconds. The caveats associated with threadDelay also apply.
--
registerDelay :: Int -> IO (TVar Bool)
registerDelay usecs = do
t <- atomically $ newTVar False
Just mgr <- getSystemEventManager
_ <- registerTimeout mgr usecs . atomically $ writeTVar t True
return t
-- | Block the current thread until data is available to read from the
-- given file descriptor.
--
-- This will throw an 'IOError' if the file descriptor was closed
-- while this thread was blocked. To safely close a file descriptor
-- that has been used with 'threadWaitRead', use 'closeFdWith'.
threadWaitRead :: Fd -> IO ()
threadWaitRead = threadWait evtRead
{-# INLINE threadWaitRead #-}
-- | Block the current thread until the given file descriptor can
-- accept data to write.
--
-- This will throw an 'IOError' if the file descriptor was closed
-- while this thread was blocked. To safely close a file descriptor
-- that has been used with 'threadWaitWrite', use 'closeFdWith'.
threadWaitWrite :: Fd -> IO ()
threadWaitWrite = threadWait evtWrite
{-# INLINE threadWaitWrite #-}
-- | Close a file descriptor in a concurrency-safe way.
--
-- Any threads that are blocked on the file descriptor via
-- 'threadWaitRead' or 'threadWaitWrite' will be unblocked by having
-- IO exceptions thrown.
closeFdWith :: (Fd -> IO ()) -- ^ Action that performs the close.
-> Fd -- ^ File descriptor to close.
-> IO ()
closeFdWith close fd = do
Just mgr <- getSystemEventManager
M.closeFd mgr close fd
threadWait :: Event -> Fd -> IO ()
threadWait evt fd = mask_ $ do
m <- newEmptyMVar
Just mgr <- getSystemEventManager
reg <- registerFd mgr (\reg e -> unregisterFd_ mgr reg >> putMVar m e) fd evt
evt' <- takeMVar m `onException` unregisterFd_ mgr reg
if evt' `eventIs` evtClose
then ioError $ errnoToIOError "threadWait" eBADF Nothing Nothing
else return ()
-- | Retrieve the system event manager.
--
-- This function always returns 'Just' the system event manager when using the
-- threaded RTS and 'Nothing' otherwise.
getSystemEventManager :: IO (Maybe EventManager)
getSystemEventManager = readIORef eventManager
foreign import ccall unsafe "getOrSetSystemEventThreadEventManagerStore"
getOrSetSystemEventThreadEventManagerStore :: Ptr a -> IO (Ptr a)
eventManager :: IORef (Maybe EventManager)
eventManager = unsafePerformIO $ do
em <- newIORef Nothing
sharedCAF em getOrSetSystemEventThreadEventManagerStore
{-# NOINLINE eventManager #-}
foreign import ccall unsafe "getOrSetSystemEventThreadIOManagerThreadStore"
getOrSetSystemEventThreadIOManagerThreadStore :: Ptr a -> IO (Ptr a)
{-# NOINLINE ioManager #-}
ioManager :: MVar (Maybe ThreadId)
ioManager = unsafePerformIO $ do
m <- newMVar Nothing
sharedCAF m getOrSetSystemEventThreadIOManagerThreadStore
ensureIOManagerIsRunning :: IO ()
ensureIOManagerIsRunning
| not threaded = return ()
| otherwise = modifyMVar_ ioManager $ \old -> do
let create = do
!mgr <- new
writeIORef eventManager $ Just mgr
!t <- forkIO $ loop mgr
labelThread t "IOManager"
return $ Just t
case old of
Nothing -> create
st@(Just t) -> do
s <- threadStatus t
case s of
ThreadFinished -> create
ThreadDied -> do
-- Sanity check: if the thread has died, there is a chance
-- that event manager is still alive. This could happend during
-- the fork, for example. In this case we should clean up
-- open pipes and everything else related to the event manager.
-- See #4449
mem <- readIORef eventManager
_ <- case mem of
Nothing -> return ()
Just em -> M.cleanup em
create
_other -> return st
foreign import ccall unsafe "rtsSupportsBoundThreads" threaded :: Bool
|
